added AI support for yaml/run

This commit is contained in:
2026-06-01 17:49:19 -03:00
parent 721a3642f3
commit 2b8e637298
26 changed files with 2885 additions and 801 deletions
+4 -4
View File
@@ -140,7 +140,7 @@ el.replaceWith(d);
def single_question(self, args, session_id):
query = " ".join(args.ask)
with console.status("[ai_status]Agent is thinking and analyzing...") as status:
with console.status("[ai_status]Agent is thinking and analyzing...[/ai_status]") as status:
result = self.app.myai.ask(query, status=status, debug=args.debug, session_id=session_id, trust=args.trust, **self.ai_overrides)
responder = result.get("responder", "engineer")
@@ -177,7 +177,7 @@ el.replaceWith(d);
if not user_query.strip(): continue
if user_query.lower() in ['exit', 'quit', 'bye', 'cancel']: break
with console.status("[ai_status]Agent is thinking...") as status:
with console.status("[ai_status]Agent is thinking...[/ai_status]") as status:
result = self.app.myai.ask(user_query, chat_history=history, status=status, debug=args.debug, trust=args.trust, session_id=session_id, **self.ai_overrides)
new_history = result.get("chat_history")
@@ -583,7 +583,7 @@ el.replaceWith(d);
if not user_query.strip(): continue
if user_query.lower() in ['exit', 'quit', 'bye', 'cancel']: break
with console.status("[ai_status]Agent is thinking...") as status:
with console.status("[ai_status]Agent is thinking...[/ai_status]") as status:
result = self.app.myai.ask(user_query, chat_history=history, status=status, debug=args.debug, trust=args.trust, session_id=session_id, **self.ai_overrides)
new_history = result.get("chat_history")
@@ -618,7 +618,7 @@ el.replaceWith(d);
</summary>
<pre><code class="python">def single_question(self, args, session_id):
query = &#34; &#34;.join(args.ask)
with console.status(&#34;[ai_status]Agent is thinking and analyzing...&#34;) as status:
with console.status(&#34;[ai_status]Agent is thinking and analyzing...[/ai_status]&#34;) as status:
result = self.app.myai.ask(query, status=status, debug=args.debug, session_id=session_id, trust=args.trust, **self.ai_overrides)
responder = result.get(&#34;responder&#34;, &#34;engineer&#34;)
+586 -7
View File
@@ -63,7 +63,12 @@ el.replaceWith(d);
def dispatch(self, args):
if len(args.data) &gt; 1:
args.action = &#34;noderun&#34;
actions = {&#34;noderun&#34;: self.node_run, &#34;generate&#34;: self.yaml_generate, &#34;run&#34;: self.yaml_run}
actions = {
&#34;noderun&#34;: self.node_run,
&#34;generate&#34;: self.yaml_generate,
&#34;generate_ai&#34;: self.ai_generate,
&#34;run&#34;: self.yaml_run
}
return actions.get(args.action)(args)
def node_run(self, args):
@@ -81,6 +86,41 @@ el.replaceWith(d);
commands = [&#34; &#34;.join(args.data[1:])]
# Check for Preflight AI simulation
if getattr(args, &#34;preflight_ai&#34;, False):
matched_node_names = [n.get(&#34;name&#34;) if isinstance(n, dict) else n for n in matched_nodes]
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Simulating execution...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[engineer][bold]Preflight AI Simulation[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
self.app.services.ai.predict_execution_results(
matched_node_names,
commands,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[engineer][bold]Preflight AI Simulation[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;engineer&#34;))
except Exception as e:
printer.error(f&#34;Preflight AI simulation failed: {e}&#34;)
sys.exit(1)
sys.exit(0)
try:
header_printed = False
@@ -118,6 +158,40 @@ el.replaceWith(d);
)
printer.run_summary(results)
# Analyze execution results if requested
if getattr(args, &#34;analyze&#34;, None) is not None:
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Analyzing execution results...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
first_chunk = False
renderer.feed(chunk)
query = args.analyze if args.analyze else &#34; &#34;.join(args.data[1:])
try:
status_context.start()
self.app.services.ai.analyze_execution_results(
results,
query=query,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;architect&#34;))
except Exception as e:
printer.error(f&#34;AI Analysis failed: {e}&#34;)
except ConnpyError as e:
printer.error(str(e))
sys.exit(1)
@@ -138,8 +212,105 @@ el.replaceWith(d);
with open(path, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
# Check preflight first before any task runs
if getattr(args, &#34;preflight_ai&#34;, False):
preflight_failed = False
for task in playbook.get(&#34;tasks&#34;, []):
name = task.get(&#34;name&#34;, &#34;Task&#34;)
nodelist = task.get(&#34;nodes&#34;, [])
commands = task.get(&#34;commands&#34;, [])
# Resolve nodes to names
try:
if isinstance(nodelist, str):
resolved_nodes = self.app.services.nodes.list_nodes(nodelist)
elif isinstance(nodelist, list):
resolved_nodes = []
for item in nodelist:
matches = self.app.services.nodes.list_nodes(item)
for m in matches:
if m not in resolved_nodes:
resolved_nodes.append(m)
else:
resolved_nodes = []
except Exception:
resolved_nodes = []
resolved_names = [n.get(&#34;name&#34;) if isinstance(n, dict) else n for n in resolved_nodes]
printer.console.print(f&#34;\n[bold]Task: {name}[/bold] (Preflight for {len(resolved_names)} nodes)&#34;)
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Simulating execution...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=f&#34;[engineer][bold]Preflight AI Simulation: {name}[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
self.app.services.ai.predict_execution_results(
resolved_names,
commands,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=f&#34;[engineer][bold]Preflight AI Simulation: {name}[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;engineer&#34;))
except Exception as e:
printer.error(f&#34;Preflight AI simulation failed for task {name}: {e}&#34;)
preflight_failed = True
if preflight_failed:
sys.exit(1)
sys.exit(0)
# Standard run
results_all = {}
for task in playbook.get(&#34;tasks&#34;, []):
self.cli_run(task)
task_res = self.cli_run(task)
if task_res:
results_all.update(task_res)
# If analyze is enabled, run analysis on accumulated results
if getattr(args, &#34;analyze&#34;, None) is not None:
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Analyzing playbook execution results...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Playbook Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
first_chunk = False
renderer.feed(chunk)
query = args.analyze if args.analyze else f&#34;Playbook: {path}&#34;
try:
status_context.start()
self.app.services.ai.analyze_execution_results(
results_all,
query=query,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Playbook Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;architect&#34;))
except Exception as e:
printer.error(f&#34;AI Analysis failed: {e}&#34;)
except Exception as e:
printer.error(f&#34;Failed to run playbook {path}: {e}&#34;)
@@ -184,6 +355,7 @@ el.replaceWith(d);
nodelist = resolved_nodes
results = {}
try:
header_printed = False
if action == &#34;run&#34;:
@@ -243,13 +415,244 @@ el.replaceWith(d);
)
# ALWAYS show the aggregate summary at the end
printer.test_summary(results)
return results
except ConnpyError as e:
printer.error(str(e))</code></pre>
printer.error(str(e))
return {}
def ai_generate(self, args):
from rich.prompt import Prompt
from rich.rule import Rule
from rich.panel import Panel
from rich.syntax import Syntax
dest_file = args.data[0]
if os.path.exists(dest_file):
printer.error(f&#34;File &#39;{dest_file}&#39; already exists.&#34;)
sys.exit(14)
chat_history = []
# Consistent layout opening matching global AI (engineer style)
from rich.markdown import Markdown
printer.console.print(Rule(style=&#34;engineer&#34;))
printer.console.print(Markdown(&#34;**Playbook Builder AI**: Welcome! Describe the automation workflow you want to design.\nType **exit** to quit.\n&#34;))
printer.console.print(Rule(style=&#34;engineer&#34;))
while True:
try:
user_prompt = Prompt.ask(&#34;[user_prompt]User[/user_prompt]&#34;)
except (KeyboardInterrupt, EOFError):
printer.console.print()
printer.warning(&#34;Operation cancelled by user.&#34;)
break
if user_prompt.strip().lower() in [&#34;exit&#34;, &#34;quit&#34;]:
printer.info(&#34;Exiting AI Assistant.&#34;)
break
if not user_prompt.strip():
continue
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Agent is thinking...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try:
status_context.stop()
except:
pass
printer.console.print(Rule(title=&#34;[engineer][bold]Playbook Builder AI[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
res = self.app.services.ai.build_playbook_chat(
user_prompt,
chat_history=chat_history,
chunk_callback=callback
)
if first_chunk:
try:
status_context.stop()
except:
pass
renderer.flush()
if not first_chunk:
printer.console.print(Rule(style=&#34;engineer&#34;))
# Update history
if res and &#34;chat_history&#34; in res:
chat_history = res[&#34;chat_history&#34;]
# Check if the agent returned a validated playbook YAML
if res and &#34;playbook_yaml&#34; in res and res[&#34;playbook_yaml&#34;]:
yaml_content = res[&#34;playbook_yaml&#34;]
printer.console.print()
printer.success(&#34;Playbook YAML successfully generated and validated.&#34;)
# Show the YAML inside a beautiful panel matching AI style (with engineer borders)
syntax = Syntax(yaml_content, &#34;yaml&#34;, theme=&#34;ansi_dark&#34;, word_wrap=True, background_color=&#34;default&#34;)
panel = Panel(syntax, title=&#34;[engineer][bold]Resulting Playbook[/bold][/engineer]&#34;, border_style=&#34;engineer&#34;, expand=False)
printer.console.print(panel)
# Ask if the user wants to save it
try:
save_confirm = Prompt.ask(
f&#34;\nDo you want to save this playbook to &#39;{dest_file}&#39;?&#34;,
choices=[&#34;y&#34;, &#34;n&#34;, &#34;run&#34;],
default=&#34;y&#34;
)
except (KeyboardInterrupt, EOFError):
printer.console.print()
printer.warning(&#34;Saving skipped.&#34;)
break
choice = save_confirm.strip().lower()
if choice in [&#34;y&#34;, &#34;yes&#34;, &#34;run&#34;]:
with open(dest_file, &#34;w&#34;) as f:
f.write(yaml_content)
printer.success(f&#34;Playbook saved successfully to &#39;{dest_file}&#39;&#34;)
if choice == &#34;run&#34;:
printer.console.print()
printer.info(&#34;Executing the saved playbook...&#34;)
self.yaml_run(args)
break
else:
printer.warning(&#34;Playbook not saved. You can continue describing changes or exit.&#34;)
except Exception as e:
printer.error(f&#34;Error in AI chat: {e}&#34;)</code></pre>
</details>
<div class="desc"></div>
<h3>Methods</h3>
<dl>
<dt id="connpy.cli.run_handler.RunHandler.ai_generate"><code class="name flex">
<span>def <span class="ident">ai_generate</span></span>(<span>self, args)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def ai_generate(self, args):
from rich.prompt import Prompt
from rich.rule import Rule
from rich.panel import Panel
from rich.syntax import Syntax
dest_file = args.data[0]
if os.path.exists(dest_file):
printer.error(f&#34;File &#39;{dest_file}&#39; already exists.&#34;)
sys.exit(14)
chat_history = []
# Consistent layout opening matching global AI (engineer style)
from rich.markdown import Markdown
printer.console.print(Rule(style=&#34;engineer&#34;))
printer.console.print(Markdown(&#34;**Playbook Builder AI**: Welcome! Describe the automation workflow you want to design.\nType **exit** to quit.\n&#34;))
printer.console.print(Rule(style=&#34;engineer&#34;))
while True:
try:
user_prompt = Prompt.ask(&#34;[user_prompt]User[/user_prompt]&#34;)
except (KeyboardInterrupt, EOFError):
printer.console.print()
printer.warning(&#34;Operation cancelled by user.&#34;)
break
if user_prompt.strip().lower() in [&#34;exit&#34;, &#34;quit&#34;]:
printer.info(&#34;Exiting AI Assistant.&#34;)
break
if not user_prompt.strip():
continue
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Agent is thinking...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try:
status_context.stop()
except:
pass
printer.console.print(Rule(title=&#34;[engineer][bold]Playbook Builder AI[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
res = self.app.services.ai.build_playbook_chat(
user_prompt,
chat_history=chat_history,
chunk_callback=callback
)
if first_chunk:
try:
status_context.stop()
except:
pass
renderer.flush()
if not first_chunk:
printer.console.print(Rule(style=&#34;engineer&#34;))
# Update history
if res and &#34;chat_history&#34; in res:
chat_history = res[&#34;chat_history&#34;]
# Check if the agent returned a validated playbook YAML
if res and &#34;playbook_yaml&#34; in res and res[&#34;playbook_yaml&#34;]:
yaml_content = res[&#34;playbook_yaml&#34;]
printer.console.print()
printer.success(&#34;Playbook YAML successfully generated and validated.&#34;)
# Show the YAML inside a beautiful panel matching AI style (with engineer borders)
syntax = Syntax(yaml_content, &#34;yaml&#34;, theme=&#34;ansi_dark&#34;, word_wrap=True, background_color=&#34;default&#34;)
panel = Panel(syntax, title=&#34;[engineer][bold]Resulting Playbook[/bold][/engineer]&#34;, border_style=&#34;engineer&#34;, expand=False)
printer.console.print(panel)
# Ask if the user wants to save it
try:
save_confirm = Prompt.ask(
f&#34;\nDo you want to save this playbook to &#39;{dest_file}&#39;?&#34;,
choices=[&#34;y&#34;, &#34;n&#34;, &#34;run&#34;],
default=&#34;y&#34;
)
except (KeyboardInterrupt, EOFError):
printer.console.print()
printer.warning(&#34;Saving skipped.&#34;)
break
choice = save_confirm.strip().lower()
if choice in [&#34;y&#34;, &#34;yes&#34;, &#34;run&#34;]:
with open(dest_file, &#34;w&#34;) as f:
f.write(yaml_content)
printer.success(f&#34;Playbook saved successfully to &#39;{dest_file}&#39;&#34;)
if choice == &#34;run&#34;:
printer.console.print()
printer.info(&#34;Executing the saved playbook...&#34;)
self.yaml_run(args)
break
else:
printer.warning(&#34;Playbook not saved. You can continue describing changes or exit.&#34;)
except Exception as e:
printer.error(f&#34;Error in AI chat: {e}&#34;)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.cli.run_handler.RunHandler.cli_run"><code class="name flex">
<span>def <span class="ident">cli_run</span></span>(<span>self, script)</span>
</code></dt>
@@ -297,6 +700,7 @@ el.replaceWith(d);
nodelist = resolved_nodes
results = {}
try:
header_printed = False
if action == &#34;run&#34;:
@@ -356,9 +760,12 @@ el.replaceWith(d);
)
# ALWAYS show the aggregate summary at the end
printer.test_summary(results)
return results
except ConnpyError as e:
printer.error(str(e))</code></pre>
printer.error(str(e))
return {}</code></pre>
</details>
<div class="desc"></div>
</dd>
@@ -373,7 +780,12 @@ el.replaceWith(d);
<pre><code class="python">def dispatch(self, args):
if len(args.data) &gt; 1:
args.action = &#34;noderun&#34;
actions = {&#34;noderun&#34;: self.node_run, &#34;generate&#34;: self.yaml_generate, &#34;run&#34;: self.yaml_run}
actions = {
&#34;noderun&#34;: self.node_run,
&#34;generate&#34;: self.yaml_generate,
&#34;generate_ai&#34;: self.ai_generate,
&#34;run&#34;: self.yaml_run
}
return actions.get(args.action)(args)</code></pre>
</details>
<div class="desc"></div>
@@ -401,6 +813,41 @@ el.replaceWith(d);
commands = [&#34; &#34;.join(args.data[1:])]
# Check for Preflight AI simulation
if getattr(args, &#34;preflight_ai&#34;, False):
matched_node_names = [n.get(&#34;name&#34;) if isinstance(n, dict) else n for n in matched_nodes]
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Simulating execution...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[engineer][bold]Preflight AI Simulation[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
self.app.services.ai.predict_execution_results(
matched_node_names,
commands,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[engineer][bold]Preflight AI Simulation[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;engineer&#34;))
except Exception as e:
printer.error(f&#34;Preflight AI simulation failed: {e}&#34;)
sys.exit(1)
sys.exit(0)
try:
header_printed = False
@@ -438,6 +885,40 @@ el.replaceWith(d);
)
printer.run_summary(results)
# Analyze execution results if requested
if getattr(args, &#34;analyze&#34;, None) is not None:
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Analyzing execution results...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
first_chunk = False
renderer.feed(chunk)
query = args.analyze if args.analyze else &#34; &#34;.join(args.data[1:])
try:
status_context.start()
self.app.services.ai.analyze_execution_results(
results,
query=query,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;architect&#34;))
except Exception as e:
printer.error(f&#34;AI Analysis failed: {e}&#34;)
except ConnpyError as e:
printer.error(str(e))
sys.exit(1)</code></pre>
@@ -478,8 +959,105 @@ el.replaceWith(d);
with open(path, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
# Check preflight first before any task runs
if getattr(args, &#34;preflight_ai&#34;, False):
preflight_failed = False
for task in playbook.get(&#34;tasks&#34;, []):
name = task.get(&#34;name&#34;, &#34;Task&#34;)
nodelist = task.get(&#34;nodes&#34;, [])
commands = task.get(&#34;commands&#34;, [])
# Resolve nodes to names
try:
if isinstance(nodelist, str):
resolved_nodes = self.app.services.nodes.list_nodes(nodelist)
elif isinstance(nodelist, list):
resolved_nodes = []
for item in nodelist:
matches = self.app.services.nodes.list_nodes(item)
for m in matches:
if m not in resolved_nodes:
resolved_nodes.append(m)
else:
resolved_nodes = []
except Exception:
resolved_nodes = []
resolved_names = [n.get(&#34;name&#34;) if isinstance(n, dict) else n for n in resolved_nodes]
printer.console.print(f&#34;\n[bold]Task: {name}[/bold] (Preflight for {len(resolved_names)} nodes)&#34;)
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Simulating execution...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=f&#34;[engineer][bold]Preflight AI Simulation: {name}[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
first_chunk = False
renderer.feed(chunk)
try:
status_context.start()
self.app.services.ai.predict_execution_results(
resolved_names,
commands,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=f&#34;[engineer][bold]Preflight AI Simulation: {name}[/bold][/engineer]&#34;, style=&#34;engineer&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;engineer&#34;))
except Exception as e:
printer.error(f&#34;Preflight AI simulation failed for task {name}: {e}&#34;)
preflight_failed = True
if preflight_failed:
sys.exit(1)
sys.exit(0)
# Standard run
results_all = {}
for task in playbook.get(&#34;tasks&#34;, []):
self.cli_run(task)
task_res = self.cli_run(task)
if task_res:
results_all.update(task_res)
# If analyze is enabled, run analysis on accumulated results
if getattr(args, &#34;analyze&#34;, None) is not None:
printer.console.print()
renderer = printer.BlockMarkdownRenderer()
first_chunk = True
status_context = printer.console.status(&#34;[ai_status]Analyzing playbook execution results...[/ai_status]&#34;)
def callback(chunk):
nonlocal first_chunk
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Playbook Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
first_chunk = False
renderer.feed(chunk)
query = args.analyze if args.analyze else f&#34;Playbook: {path}&#34;
try:
status_context.start()
self.app.services.ai.analyze_execution_results(
results_all,
query=query,
chunk_callback=callback
)
if first_chunk:
try: status_context.stop()
except: pass
printer.console.print(Rule(title=&#34;[architect][bold]Network Architect AI Playbook Analysis[/bold][/architect]&#34;, style=&#34;architect&#34;))
renderer.flush()
printer.console.print(Rule(style=&#34;architect&#34;))
except Exception as e:
printer.error(f&#34;AI Analysis failed: {e}&#34;)
except Exception as e:
printer.error(f&#34;Failed to run playbook {path}: {e}&#34;)
@@ -506,7 +1084,8 @@ el.replaceWith(d);
<ul>
<li>
<h4><code><a title="connpy.cli.run_handler.RunHandler" href="#connpy.cli.run_handler.RunHandler">RunHandler</a></code></h4>
<ul class="">
<ul class="two-column">
<li><code><a title="connpy.cli.run_handler.RunHandler.ai_generate" href="#connpy.cli.run_handler.RunHandler.ai_generate">ai_generate</a></code></li>
<li><code><a title="connpy.cli.run_handler.RunHandler.cli_run" href="#connpy.cli.run_handler.RunHandler.cli_run">cli_run</a></code></li>
<li><code><a title="connpy.cli.run_handler.RunHandler.dispatch" href="#connpy.cli.run_handler.RunHandler.dispatch">dispatch</a></code></li>
<li><code><a title="connpy.cli.run_handler.RunHandler.node_run" href="#connpy.cli.run_handler.RunHandler.node_run">node_run</a></code></li>
+296 -100
View File
@@ -100,6 +100,21 @@ el.replaceWith(d);
request_deserializer=connpy__pb2.StringRequest.FromString,
response_serializer=connpy__pb2.StructResponse.SerializeToString,
),
&#39;build_playbook_chat&#39;: grpc.stream_stream_rpc_method_handler(
servicer.build_playbook_chat,
request_deserializer=connpy__pb2.AskRequest.FromString,
response_serializer=connpy__pb2.AIResponse.SerializeToString,
),
&#39;analyze_execution_results&#39;: grpc.unary_stream_rpc_method_handler(
servicer.analyze_execution_results,
request_deserializer=connpy__pb2.AnalyzeRequest.FromString,
response_serializer=connpy__pb2.AIResponse.SerializeToString,
),
&#39;predict_execution_results&#39;: grpc.unary_stream_rpc_method_handler(
servicer.predict_execution_results,
request_deserializer=connpy__pb2.PreflightRequest.FromString,
response_serializer=connpy__pb2.AIResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
&#39;connpy.AIService&#39;, rpc_method_handlers)
@@ -209,11 +224,6 @@ el.replaceWith(d);
request_deserializer=connpy__pb2.ScriptRequest.FromString,
response_serializer=connpy__pb2.StructResponse.SerializeToString,
),
&#39;run_yaml_playbook&#39;: grpc.unary_unary_rpc_method_handler(
servicer.run_yaml_playbook,
request_deserializer=connpy__pb2.ScriptRequest.FromString,
response_serializer=connpy__pb2.StructResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
&#39;connpy.ExecutionService&#39;, rpc_method_handlers)
@@ -739,11 +749,129 @@ el.replaceWith(d);
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def build_playbook_chat(request_iterator,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.stream_stream(
request_iterator,
target,
&#39;/connpy.AIService/build_playbook_chat&#39;,
connpy__pb2.AskRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def analyze_execution_results(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
&#39;/connpy.AIService/analyze_execution_results&#39;,
connpy__pb2.AnalyzeRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def predict_execution_results(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
&#39;/connpy.AIService/predict_execution_results&#39;,
connpy__pb2.PreflightRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
<h3>Static methods</h3>
<dl>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIService.analyze_execution_results"><code class="name flex">
<span>def <span class="ident">analyze_execution_results</span></span>(<span>request,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@staticmethod
def analyze_execution_results(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
&#39;/connpy.AIService/analyze_execution_results&#39;,
connpy__pb2.AnalyzeRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIService.ask"><code class="name flex">
<span>def <span class="ident">ask</span></span>(<span>request_iterator,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
@@ -818,6 +946,43 @@ def ask_copilot(request,
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIService.build_playbook_chat"><code class="name flex">
<span>def <span class="ident">build_playbook_chat</span></span>(<span>request_iterator,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@staticmethod
def build_playbook_chat(request_iterator,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.stream_stream(
request_iterator,
target,
&#39;/connpy.AIService/build_playbook_chat&#39;,
connpy__pb2.AskRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIService.configure_mcp"><code class="name flex">
<span>def <span class="ident">configure_mcp</span></span>(<span>request,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
@@ -1077,6 +1242,43 @@ def load_session_data(request,
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIService.predict_execution_results"><code class="name flex">
<span>def <span class="ident">predict_execution_results</span></span>(<span>request,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@staticmethod
def predict_execution_results(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
&#39;/connpy.AIService/predict_execution_results&#39;,
connpy__pb2.PreflightRequest.SerializeToString,
connpy__pb2.AIResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"></div>
</dd>
</dl>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer"><code class="flex name class">
@@ -1139,6 +1341,24 @@ def load_session_data(request,
raise NotImplementedError(&#39;Method not implemented!&#39;)
def load_session_data(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)
def build_playbook_chat(self, request_iterator, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)
def analyze_execution_results(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)
def predict_execution_results(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
@@ -1151,6 +1371,22 @@ def load_session_data(request,
</ul>
<h3>Methods</h3>
<dl>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.analyze_execution_results"><code class="name flex">
<span>def <span class="ident">analyze_execution_results</span></span>(<span>self, request, context)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def analyze_execution_results(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask"><code class="name flex">
<span>def <span class="ident">ask</span></span>(<span>self, request_iterator, context)</span>
</code></dt>
@@ -1183,6 +1419,22 @@ def load_session_data(request,
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.build_playbook_chat"><code class="name flex">
<span>def <span class="ident">build_playbook_chat</span></span>(<span>self, request_iterator, context)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def build_playbook_chat(self, request_iterator, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_mcp"><code class="name flex">
<span>def <span class="ident">configure_mcp</span></span>(<span>self, request, context)</span>
</code></dt>
@@ -1295,6 +1547,22 @@ def load_session_data(request,
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.predict_execution_results"><code class="name flex">
<span>def <span class="ident">predict_execution_results</span></span>(<span>self, request, context)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def predict_execution_results(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
</dl>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.AIServiceStub"><code class="flex name class">
@@ -1359,6 +1627,21 @@ def load_session_data(request,
&#39;/connpy.AIService/load_session_data&#39;,
request_serializer=connpy__pb2.StringRequest.SerializeToString,
response_deserializer=connpy__pb2.StructResponse.FromString,
_registered_method=True)
self.build_playbook_chat = channel.stream_stream(
&#39;/connpy.AIService/build_playbook_chat&#39;,
request_serializer=connpy__pb2.AskRequest.SerializeToString,
response_deserializer=connpy__pb2.AIResponse.FromString,
_registered_method=True)
self.analyze_execution_results = channel.unary_stream(
&#39;/connpy.AIService/analyze_execution_results&#39;,
request_serializer=connpy__pb2.AnalyzeRequest.SerializeToString,
response_deserializer=connpy__pb2.AIResponse.FromString,
_registered_method=True)
self.predict_execution_results = channel.unary_stream(
&#39;/connpy.AIService/predict_execution_results&#39;,
request_serializer=connpy__pb2.PreflightRequest.SerializeToString,
response_deserializer=connpy__pb2.AIResponse.FromString,
_registered_method=True)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p>
@@ -2313,33 +2596,6 @@ def update_setting(request,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def run_yaml_playbook(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
&#39;/connpy.ExecutionService/run_yaml_playbook&#39;,
connpy__pb2.ScriptRequest.SerializeToString,
connpy__pb2.StructResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
@@ -2419,43 +2675,6 @@ def run_commands(request,
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_yaml_playbook"><code class="name flex">
<span>def <span class="ident">run_yaml_playbook</span></span>(<span>request,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@staticmethod
def run_yaml_playbook(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
&#39;/connpy.ExecutionService/run_yaml_playbook&#39;,
connpy__pb2.ScriptRequest.SerializeToString,
connpy__pb2.StructResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.test_commands"><code class="name flex">
<span>def <span class="ident">test_commands</span></span>(<span>request,<br>target,<br>options=(),<br>channel_credentials=None,<br>call_credentials=None,<br>insecure=False,<br>compression=None,<br>wait_for_ready=None,<br>timeout=None,<br>metadata=None)</span>
</code></dt>
@@ -2519,12 +2738,6 @@ def test_commands(request,
raise NotImplementedError(&#39;Method not implemented!&#39;)
def run_cli_script(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)
def run_yaml_playbook(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
@@ -2569,22 +2782,6 @@ def test_commands(request,
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_yaml_playbook"><code class="name flex">
<span>def <span class="ident">run_yaml_playbook</span></span>(<span>self, request, context)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def run_yaml_playbook(self, request, context):
&#34;&#34;&#34;Missing associated documentation comment in .proto file.&#34;&#34;&#34;
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details(&#39;Method not implemented!&#39;)
raise NotImplementedError(&#39;Method not implemented!&#39;)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
</dd>
<dt id="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.test_commands"><code class="name flex">
<span>def <span class="ident">test_commands</span></span>(<span>self, request, context)</span>
</code></dt>
@@ -2635,11 +2832,6 @@ def test_commands(request,
&#39;/connpy.ExecutionService/run_cli_script&#39;,
request_serializer=connpy__pb2.ScriptRequest.SerializeToString,
response_deserializer=connpy__pb2.StructResponse.FromString,
_registered_method=True)
self.run_yaml_playbook = channel.unary_unary(
&#39;/connpy.ExecutionService/run_yaml_playbook&#39;,
request_serializer=connpy__pb2.ScriptRequest.SerializeToString,
response_deserializer=connpy__pb2.StructResponse.FromString,
_registered_method=True)</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p>
@@ -6089,9 +6281,11 @@ def stop_api(request,
<ul>
<li>
<h4><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService">AIService</a></code></h4>
<ul class="two-column">
<ul class="">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.analyze_execution_results" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.ask" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.ask">ask</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.ask_copilot" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.ask_copilot">ask_copilot</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.build_playbook_chat" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.configure_mcp" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.configure_provider" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.confirm" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.confirm">confirm</a></code></li>
@@ -6099,13 +6293,16 @@ def stop_api(request,
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.list_mcp_servers" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.list_sessions" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.load_session_data" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIService.predict_execution_results" href="#connpy.grpc_layer.connpy_pb2_grpc.AIService.predict_execution_results">predict_execution_results</a></code></li>
</ul>
</li>
<li>
<h4><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer">AIServiceServicer</a></code></h4>
<ul class="two-column">
<ul class="">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.analyze_execution_results" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask">ask</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask_copilot" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask_copilot">ask_copilot</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.build_playbook_chat" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_mcp" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_provider" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.confirm" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.confirm">confirm</a></code></li>
@@ -6113,6 +6310,7 @@ def stop_api(request,
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_mcp_servers" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_sessions" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.load_session_data" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.predict_execution_results" href="#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.predict_execution_results">predict_execution_results</a></code></li>
</ul>
</li>
<li>
@@ -6165,7 +6363,6 @@ def stop_api(request,
<ul class="">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_cli_script" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_commands" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_yaml_playbook" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.test_commands" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionService.test_commands">test_commands</a></code></li>
</ul>
</li>
@@ -6174,7 +6371,6 @@ def stop_api(request,
<ul class="">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_cli_script" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_commands" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_yaml_playbook" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.test_commands" href="#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.test_commands">test_commands</a></code></li>
</ul>
</li>
+89 -22
View File
@@ -174,12 +174,10 @@ el.replaceWith(d);
def service(self):
return self._get_provider().ai
@handle_errors
def ask(self, request_iterator, context):
def _handle_chat_stream(self, request_iterator, context, service_method):
import queue
import threading
ai_service = self.service
chunk_queue = queue.Queue()
request_queue = queue.Queue()
bridge = None
@@ -197,21 +195,28 @@ el.replaceWith(d);
nonlocal history, bridge, agent_instance
try:
# Run the AI interaction (this blocks this specific thread)
res = ai_service.ask(
input_text,
chat_history=history if history else None,
session_id=session_id,
debug=debug,
status=bridge,
console=bridge,
confirm_handler=bridge.confirm,
chunk_callback=callback,
trust=trust,
**overrides
)
if getattr(service_method, &#34;__name__&#34;, None) == &#34;build_playbook_chat&#34;:
res = service_method(
input_text,
chat_history=history if history else None,
status=bridge,
chunk_callback=callback
)
else:
res = service_method(
input_text,
chat_history=history if history else None,
session_id=session_id,
debug=debug,
status=bridge,
confirm_handler=bridge.confirm,
chunk_callback=callback,
trust=trust,
**overrides
)
# Update history for next message
if &#34;chat_history&#34; in res:
if res and &#34;chat_history&#34; in res:
history = res[&#34;chat_history&#34;]
# Send final chunk marker
@@ -305,6 +310,71 @@ el.replaceWith(d);
elif msg_type == &#34;final_mark&#34;:
yield connpy_pb2.AIResponse(is_final=True, full_result=to_struct(val))
def _handle_unary_stream(self, service_method, *args, **kwargs):
import queue
import threading
chunk_queue = queue.Queue()
bridge = StatusBridge(chunk_queue, is_web=False)
def callback(chunk):
chunk_queue.put((&#34;text&#34;, chunk))
def _worker():
try:
res = service_method(*args, chunk_callback=callback, status=bridge, **kwargs)
chunk_queue.put((&#34;final_mark&#34;, res))
except Exception as e:
import traceback
print(f&#34;gRPC Unary Stream error: {e}&#34;)
traceback.print_exc()
chunk_queue.put((&#34;status&#34;, f&#34;Error: {str(e)}&#34;))
chunk_queue.put((&#34;final_mark&#34;, {&#34;response&#34;: f&#34;Error: {str(e)}&#34;, &#34;error&#34;: True}))
finally:
chunk_queue.put((None, None))
threading.Thread(target=_worker, daemon=True).start()
while True:
item = chunk_queue.get()
if item == (None, None):
break
msg_type, val = item
if msg_type == &#34;text&#34;:
yield connpy_pb2.AIResponse(text_chunk=val, is_final=False)
elif msg_type == &#34;status&#34;:
clean_val = val.replace(&#34;[ai_status]&#34;, &#34;&#34;).replace(&#34;[/ai_status]&#34;, &#34;&#34;)
yield connpy_pb2.AIResponse(status_update=clean_val, is_final=False)
elif msg_type == &#34;debug&#34;:
yield connpy_pb2.AIResponse(debug_message=val, is_final=False)
elif msg_type == &#34;important&#34;:
yield connpy_pb2.AIResponse(important_message=val, is_final=False)
elif msg_type == &#34;confirm&#34;:
yield connpy_pb2.AIResponse(status_update=val, requires_confirmation=True, is_final=False)
elif msg_type == &#34;final_mark&#34;:
yield connpy_pb2.AIResponse(is_final=True, full_result=to_struct(val))
@handle_errors
def ask(self, request_iterator, context):
yield from self._handle_chat_stream(request_iterator, context, self.service.ask)
@handle_errors
def build_playbook_chat(self, request_iterator, context):
yield from self._handle_chat_stream(request_iterator, context, self.service.build_playbook_chat)
@handle_errors
def analyze_execution_results(self, request, context):
results = from_struct(request.results)
query = request.query if request.query else None
yield from self._handle_unary_stream(self.service.analyze_execution_results, results, query=query)
@handle_errors
def predict_execution_results(self, request, context):
target_nodes = list(request.target_nodes)
commands = list(request.commands)
yield from self._handle_unary_stream(self.service.predict_execution_results, target_nodes, commands)
@handle_errors
def confirm(self, request, context):
res = self.service.confirm(request.value)
@@ -386,8 +456,10 @@ def service(self):
<ul class="hlist">
<li><code><b><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer">AIServiceServicer</a></b></code>:
<ul class="hlist">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.analyze_execution_results" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask">ask</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask_copilot" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.ask_copilot">ask_copilot</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.build_playbook_chat" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_mcp" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_provider" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.confirm" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.confirm">confirm</a></code></li>
@@ -395,6 +467,7 @@ def service(self):
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_mcp_servers" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_sessions" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.load_session_data" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.predict_execution_results" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.AIServiceServicer.predict_execution_results">predict_execution_results</a></code></li>
</ul>
</li>
</ul>
@@ -855,11 +928,6 @@ def service(self):
@handle_errors
def run_cli_script(self, request, context):
res = self.service.run_cli_script(request.param1, request.param2, request.parallel)
return connpy_pb2.StructResponse(data=to_struct(res))
@handle_errors
def run_yaml_playbook(self, request, context):
res = self.service.run_yaml_playbook(request.param1, request.parallel)
return connpy_pb2.StructResponse(data=to_struct(res))</code></pre>
</details>
<div class="desc"><p>Missing associated documentation comment in .proto file.</p></div>
@@ -888,7 +956,6 @@ def service(self):
<ul class="hlist">
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_cli_script" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_commands" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_yaml_playbook" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.test_commands" href="connpy_pb2_grpc.html#connpy.grpc_layer.connpy_pb2_grpc.ExecutionServiceServicer.test_commands">test_commands</a></code></li>
</ul>
</li>
+175 -230
View File
@@ -99,8 +99,7 @@ el.replaceWith(d);
self.stub = connpy_pb2_grpc.AIServiceStub(channel)
self.remote_host = remote_host
@handle_errors
def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debug=False, status=None, **overrides):
def _ai_chat_stream(self, stub_method, input_text, dryrun=False, chat_history=None, session_id=None, debug=False, status=None, chunk_callback=None, **overrides):
import queue
from rich.prompt import Prompt
from rich.text import Text
@@ -135,7 +134,7 @@ el.replaceWith(d);
if req is None: break
yield req
responses = self.stub.ask(request_generator())
responses = stub_method(request_generator())
full_content = &#34;&#34;
header_printed = False
@@ -234,26 +233,32 @@ el.replaceWith(d);
try: status.stop()
except: pass
from rich.console import Console as RichConsole
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
# Print header on first chunk
alias = &#34;architect&#34; if current_responder == &#34;architect&#34; else &#34;engineer&#34;
role_label = &#34;Network Architect&#34; if current_responder == &#34;architect&#34; else &#34;Network Engineer&#34;
stable_console.print(Rule(f&#34;[bold {alias}]{role_label}[/bold {alias}]&#34;, style=alias))
header_printed = True
# Initialize parser
md_parser = IncrementalMarkdownParser(console=stable_console)
if chunk_callback:
header_printed = True
else:
from rich.console import Console as RichConsole
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
# Print header on first chunk
alias = &#34;architect&#34; if current_responder == &#34;architect&#34; else &#34;engineer&#34;
role_label = &#34;Network Architect&#34; if current_responder == &#34;architect&#34; else &#34;Network Engineer&#34;
stable_console.print(Rule(f&#34;[bold {alias}]{role_label}[/bold {alias}]&#34;, style=alias))
header_printed = True
# Initialize parser
md_parser = IncrementalMarkdownParser(console=stable_console)
full_content += response.text_chunk
md_parser.feed(response.text_chunk)
if chunk_callback:
chunk_callback(response.text_chunk)
elif md_parser:
md_parser.feed(response.text_chunk)
continue
if response.is_final:
if header_printed:
if not chunk_callback and header_printed:
from rich.rule import Rule
md_parser.flush()
@@ -262,12 +267,8 @@ el.replaceWith(d);
except: pass
final_result = from_struct(response.full_result)
responder = final_result.get(&#34;responder&#34;, &#34;engineer&#34;)
alias = &#34;architect&#34; if responder == &#34;architect&#34; else &#34;engineer&#34;
role_label = &#34;Network Architect&#34; if responder == &#34;architect&#34; else &#34;Network Engineer&#34;
title = f&#34;[bold {alias}]{role_label}[/bold {alias}]&#34;
if header_printed:
if not chunk_callback and header_printed:
from rich.console import Console as RichConsole
from ..printer import connpy_theme, get_original_stdout
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
@@ -286,6 +287,104 @@ el.replaceWith(d);
return final_result
@handle_errors
def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debug=False, status=None, **overrides):
return self._ai_chat_stream(self.stub.ask, input_text, dryrun=dryrun, chat_history=chat_history, session_id=session_id, debug=debug, status=status, **overrides)
@handle_errors
def build_playbook_chat(self, user_input, chat_history=None, status=None, chunk_callback=None):
return self._ai_chat_stream(self.stub.build_playbook_chat, user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)
def _process_unary_stream(self, responses, status=None, chunk_callback=None):
full_content = &#34;&#34;
header_printed = False
final_result = {&#34;response&#34;: &#34;&#34;, &#34;chat_history&#34;: []}
md_parser = None
try:
for response in responses:
if response.status_update:
if status:
status.update(response.status_update)
continue
if response.important_message:
if status:
try: status.stop()
except: pass
printer.console.print(Text.from_ansi(response.important_message))
if status:
try: status.start()
except: pass
continue
if not response.is_final:
if response.text_chunk:
if not header_printed:
if status:
try: status.stop()
except: pass
if chunk_callback:
header_printed = True
else:
from rich.console import Console as RichConsole
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
# Print default header
stable_console.print(Rule(&#34;[bold engineer]AI Analysis[/bold engineer]&#34;, style=&#34;engineer&#34;))
header_printed = True
md_parser = IncrementalMarkdownParser(console=stable_console)
full_content += response.text_chunk
if chunk_callback:
chunk_callback(response.text_chunk)
elif md_parser:
md_parser.feed(response.text_chunk)
continue
if response.is_final:
if md_parser:
md_parser.flush()
if status:
try: status.stop()
except: pass
final_result = from_struct(response.full_result)
if md_parser:
from rich.console import Console as RichConsole
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
stable_console.print(Rule(style=&#34;engineer&#34;))
break
except Exception as e:
if isinstance(e, grpc.RpcError):
raise
printer.warning(f&#34;Stream interrupted: {e}&#34;)
if full_content:
final_result[&#34;streamed&#34;] = True
return final_result
@handle_errors
def analyze_execution_results(self, results, query=None, status=None, chunk_callback=None):
req = connpy_pb2.AnalyzeRequest(query=query or &#34;&#34;)
req.results.CopyFrom(to_struct(results))
responses = self.stub.analyze_execution_results(req)
return self._process_unary_stream(responses, status, chunk_callback)
@handle_errors
def predict_execution_results(self, target_nodes, commands, status=None, chunk_callback=None):
req = connpy_pb2.PreflightRequest(target_nodes=target_nodes, commands=commands)
responses = self.stub.predict_execution_results(req)
return self._process_unary_stream(responses, status, chunk_callback)
@handle_errors
def confirm(self, input_text, console=None):
return self.stub.confirm(connpy_pb2.StringRequest(value=input_text)).value
@@ -333,6 +432,23 @@ el.replaceWith(d);
<div class="desc"></div>
<h3>Methods</h3>
<dl>
<dt id="connpy.grpc_layer.stubs.AIStub.analyze_execution_results"><code class="name flex">
<span>def <span class="ident">analyze_execution_results</span></span>(<span>self, results, query=None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@handle_errors
def analyze_execution_results(self, results, query=None, status=None, chunk_callback=None):
req = connpy_pb2.AnalyzeRequest(query=query or &#34;&#34;)
req.results.CopyFrom(to_struct(results))
responses = self.stub.analyze_execution_results(req)
return self._process_unary_stream(responses, status, chunk_callback)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.stubs.AIStub.ask"><code class="name flex">
<span>def <span class="ident">ask</span></span>(<span>self,<br>input_text,<br>dryrun=False,<br>chat_history=None,<br>session_id=None,<br>debug=False,<br>status=None,<br>**overrides)</span>
</code></dt>
@@ -343,190 +459,21 @@ el.replaceWith(d);
</summary>
<pre><code class="python">@handle_errors
def ask(self, input_text, dryrun=False, chat_history=None, session_id=None, debug=False, status=None, **overrides):
import queue
from rich.prompt import Prompt
from rich.text import Text
from rich.panel import Panel
from rich.markdown import Markdown
req_queue = queue.Queue()
initial_req = connpy_pb2.AskRequest(
input_text=input_text,
dryrun=dryrun,
session_id=session_id or &#34;&#34;,
debug=debug,
engineer_model=overrides.get(&#34;engineer_model&#34;, &#34;&#34;),
engineer_api_key=overrides.get(&#34;engineer_api_key&#34;, &#34;&#34;),
architect_model=overrides.get(&#34;architect_model&#34;, &#34;&#34;),
architect_api_key=overrides.get(&#34;architect_api_key&#34;, &#34;&#34;),
trust=overrides.get(&#34;trust&#34;, False)
)
if chat_history is not None:
initial_req.chat_history.CopyFrom(to_value(chat_history))
if &#34;engineer_auth&#34; in overrides and overrides[&#34;engineer_auth&#34;]:
initial_req.engineer_auth.CopyFrom(to_struct(overrides[&#34;engineer_auth&#34;]))
if &#34;architect_auth&#34; in overrides and overrides[&#34;architect_auth&#34;]:
initial_req.architect_auth.CopyFrom(to_struct(overrides[&#34;architect_auth&#34;]))
req_queue.put(initial_req)
def request_generator():
while True:
req = req_queue.get()
if req is None: break
yield req
responses = self.stub.ask(request_generator())
full_content = &#34;&#34;
header_printed = False
current_responder = &#34;engineer&#34;
final_result = {&#34;response&#34;: &#34;&#34;, &#34;chat_history&#34;: []}
# Background thread to pull responses from gRPC into a local queue
# This prevents KeyboardInterrupt from corrupting the gRPC iterator state
response_queue = queue.Queue()
def pull_responses():
try:
for response in responses:
response_queue.put((&#34;data&#34;, response))
except Exception as e:
response_queue.put((&#34;error&#34;, e))
finally:
response_queue.put((None, None))
threading.Thread(target=pull_responses, daemon=True).start()
try:
while True:
try:
# BLOCKING GET from local queue (interruptible by signal)
msg_type, response = response_queue.get()
except KeyboardInterrupt:
# Signal interruption to the server
if status:
status.update(&#34;[error]Interrupted! Closing pending tasks...&#34;)
# Send the interrupt signal to the server
req_queue.put(connpy_pb2.AskRequest(interrupt=True))
# CONTINUE the loop to receive remaining data and summary from the queue
continue
if msg_type is None: # Sentinel
break
if msg_type == &#34;error&#34;:
# Re-raise or handle gRPC error from background thread
if isinstance(response, grpc.RpcError):
raise response
printer.warning(f&#34;Stream interrupted: {response}&#34;)
break
if response.status_update:
if response.status_update.startswith(&#34;__RESPONDER__:&#34;):
current_responder = response.status_update.split(&#34;:&#34;)[1].lower()
continue
if response.requires_confirmation:
if status: status.stop()
# Show prompt and wait for answer
prompt_text = Text.from_ansi(response.status_update)
ans = Prompt.ask(prompt_text)
if status:
status.update(&#34;[ai_status]Agent: Resuming...&#34;)
status.start()
req_queue.put(connpy_pb2.AskRequest(confirmation_answer=ans))
continue
if status:
status.update(response.status_update)
continue
if response.debug_message:
if debug:
if status:
try: status.stop()
except: pass
printer.console.print(Text.from_ansi(response.debug_message))
if status:
try: status.start()
except: pass
continue
if response.important_message:
if status:
try: status.stop()
except: pass
printer.console.print(Text.from_ansi(response.important_message))
if status:
try: status.start()
except: pass
continue
if not response.is_final:
if response.text_chunk:
if not header_printed:
if status:
try: status.stop()
except: pass
from rich.console import Console as RichConsole
from rich.rule import Rule
from ..printer import connpy_theme, get_original_stdout, IncrementalMarkdownParser
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
# Print header on first chunk
alias = &#34;architect&#34; if current_responder == &#34;architect&#34; else &#34;engineer&#34;
role_label = &#34;Network Architect&#34; if current_responder == &#34;architect&#34; else &#34;Network Engineer&#34;
stable_console.print(Rule(f&#34;[bold {alias}]{role_label}[/bold {alias}]&#34;, style=alias))
header_printed = True
# Initialize parser
md_parser = IncrementalMarkdownParser(console=stable_console)
full_content += response.text_chunk
md_parser.feed(response.text_chunk)
continue
if response.is_final:
if header_printed:
from rich.rule import Rule
md_parser.flush()
if status:
try: status.stop()
except: pass
final_result = from_struct(response.full_result)
responder = final_result.get(&#34;responder&#34;, &#34;engineer&#34;)
alias = &#34;architect&#34; if responder == &#34;architect&#34; else &#34;engineer&#34;
role_label = &#34;Network Architect&#34; if responder == &#34;architect&#34; else &#34;Network Engineer&#34;
title = f&#34;[bold {alias}]{role_label}[/bold {alias}]&#34;
if header_printed:
from rich.console import Console as RichConsole
from ..printer import connpy_theme, get_original_stdout
stable_console = RichConsole(theme=connpy_theme, file=get_original_stdout())
stable_console.print(Rule(style=alias))
break
except Exception as e:
# Check if it was a gRPC error that we should let handle_errors catch
if isinstance(e, grpc.RpcError):
raise
printer.warning(f&#34;Stream interrupted: {e}&#34;)
finally:
req_queue.put(None)
if full_content:
final_result[&#34;streamed&#34;] = True
return final_result</code></pre>
return self._ai_chat_stream(self.stub.ask, input_text, dryrun=dryrun, chat_history=chat_history, session_id=session_id, debug=debug, status=status, **overrides)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.stubs.AIStub.build_playbook_chat"><code class="name flex">
<span>def <span class="ident">build_playbook_chat</span></span>(<span>self, user_input, chat_history=None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@handle_errors
def build_playbook_chat(self, user_input, chat_history=None, status=None, chunk_callback=None):
return self._ai_chat_stream(self.stub.build_playbook_chat, user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"></div>
</dd>
@@ -644,6 +591,22 @@ def load_session_data(self, session_id):
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.stubs.AIStub.predict_execution_results"><code class="name flex">
<span>def <span class="ident">predict_execution_results</span></span>(<span>self, target_nodes, commands, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@handle_errors
def predict_execution_results(self, target_nodes, commands, status=None, chunk_callback=None):
req = connpy_pb2.PreflightRequest(target_nodes=target_nodes, commands=commands)
responses = self.stub.predict_execution_results(req)
return self._process_unary_stream(responses, status, chunk_callback)</code></pre>
</details>
<div class="desc"></div>
</dd>
</dl>
</dd>
<dt id="connpy.grpc_layer.stubs.AuthClientInterceptor"><code class="flex name class">
@@ -1124,12 +1087,7 @@ def update_setting(self, key, value):
@handle_errors
def run_cli_script(self, nodes_filter, script_path, parallel=10):
req = connpy_pb2.ScriptRequest(param1=nodes_filter, param2=script_path, parallel=parallel)
return from_struct(self.stub.run_cli_script(req).data)
@handle_errors
def run_yaml_playbook(self, playbook_path, parallel=10):
req = connpy_pb2.ScriptRequest(param1=playbook_path, parallel=parallel)
return from_struct(self.stub.run_yaml_playbook(req).data)</code></pre>
return from_struct(self.stub.run_cli_script(req).data)</code></pre>
</details>
<div class="desc"></div>
<h3>Methods</h3>
@@ -1187,21 +1145,6 @@ def run_commands(self, nodes_filter, commands, variables=None, parallel=10, time
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.stubs.ExecutionStub.run_yaml_playbook"><code class="name flex">
<span>def <span class="ident">run_yaml_playbook</span></span>(<span>self, playbook_path, parallel=10)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">@handle_errors
def run_yaml_playbook(self, playbook_path, parallel=10):
req = connpy_pb2.ScriptRequest(param1=playbook_path, parallel=parallel)
return from_struct(self.stub.run_yaml_playbook(req).data)</code></pre>
</details>
<div class="desc"></div>
</dd>
<dt id="connpy.grpc_layer.stubs.ExecutionStub.test_commands"><code class="name flex">
<span>def <span class="ident">test_commands</span></span>(<span>self,<br>nodes_filter,<br>commands,<br>expected,<br>variables=None,<br>parallel=10,<br>timeout=10,<br>prompt=None,<br>**kwargs)</span>
</code></dt>
@@ -2815,8 +2758,10 @@ def stop_api(self):
<ul>
<li>
<h4><code><a title="connpy.grpc_layer.stubs.AIStub" href="#connpy.grpc_layer.stubs.AIStub">AIStub</a></code></h4>
<ul class="two-column">
<ul class="">
<li><code><a title="connpy.grpc_layer.stubs.AIStub.analyze_execution_results" href="#connpy.grpc_layer.stubs.AIStub.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.ask" href="#connpy.grpc_layer.stubs.AIStub.ask">ask</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.build_playbook_chat" href="#connpy.grpc_layer.stubs.AIStub.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.configure_mcp" href="#connpy.grpc_layer.stubs.AIStub.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.configure_provider" href="#connpy.grpc_layer.stubs.AIStub.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.confirm" href="#connpy.grpc_layer.stubs.AIStub.confirm">confirm</a></code></li>
@@ -2824,6 +2769,7 @@ def stop_api(self):
<li><code><a title="connpy.grpc_layer.stubs.AIStub.list_mcp_servers" href="#connpy.grpc_layer.stubs.AIStub.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.list_sessions" href="#connpy.grpc_layer.stubs.AIStub.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.load_session_data" href="#connpy.grpc_layer.stubs.AIStub.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.AIStub.predict_execution_results" href="#connpy.grpc_layer.stubs.AIStub.predict_execution_results">predict_execution_results</a></code></li>
</ul>
</li>
<li>
@@ -2857,7 +2803,6 @@ def stop_api(self):
<ul class="">
<li><code><a title="connpy.grpc_layer.stubs.ExecutionStub.run_cli_script" href="#connpy.grpc_layer.stubs.ExecutionStub.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.ExecutionStub.run_commands" href="#connpy.grpc_layer.stubs.ExecutionStub.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.ExecutionStub.run_yaml_playbook" href="#connpy.grpc_layer.stubs.ExecutionStub.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.grpc_layer.stubs.ExecutionStub.test_commands" href="#connpy.grpc_layer.stubs.ExecutionStub.test_commands">test_commands</a></code></li>
</ul>
</li>
+68 -46
View File
@@ -185,6 +185,8 @@ response = myai.ask(&quot;What is the status of the BGP neighbors in the office?
</code></pre>
<hr>
<p><em>For detailed developer notes and plugin hooks documentation, see the <a href="https://fluzzi.github.io/connpy/">Documentation</a>.</em></p>
<h2 id="license">📜 License</h2>
<p><a href="LICENSE">PolyForm Noncommercial 1.0.0</a></p>
</section>
<section>
<h2 class="section-title" id="header-submodules">Sub-modules</h2>
@@ -644,6 +646,7 @@ class ai:
self.confirm_handler = confirm_handler or self._local_confirm_handler
self.trusted_session = trust # Trust mode for the entire session
self.interrupted = False
self.one_shot = kwargs.get(&#34;one_shot&#34;, False)
# 1. Cargar configuración genérica con herencia/merge global
@@ -815,10 +818,13 @@ class ai:
@property
def architect_system_prompt(self):
&#34;&#34;&#34;Build architect system prompt with plugin extensions.&#34;&#34;&#34;
prompt = self._architect_base_prompt
if getattr(self, &#34;one_shot&#34;, False):
prompt += &#34;\n\nCRITICAL 1-SHOT DIAGNOSTICS DIRECTIVE:\nYou are running in a 1-shot offline diagnostics mode. There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately and directly to the user. Do not suggest or attempt to delegate/return control to the engineer.&#34;
if self.architect_prompt_extensions:
extensions = &#34;\n&#34;.join(self.architect_prompt_extensions)
return self._architect_base_prompt + f&#34;\n\nPlugin Capabilities:\n{extensions}&#34;
return self._architect_base_prompt
return prompt + f&#34;\n\nPlugin Capabilities:\n{extensions}&#34;
return prompt
def register_ai_tool(self, tool_definition, handler, target=&#34;engineer&#34;, engineer_prompt=None, architect_prompt=None, status_formatter=None):
&#34;&#34;&#34;Register an external tool for the AI system.
@@ -1295,13 +1301,11 @@ class ai:
if self.interrupted:
raise KeyboardInterrupt
# Soft limit warning
if iteration == self.soft_limit_iterations and not soft_limit_warned:
self.console.print(f&#34;[warning]⚠ Engineer has performed {iteration} steps. This is taking longer than expected.[/warning]&#34;)
self.console.print(f&#34;[warning] You can press Ctrl+C to interrupt and get a summary.[/warning]&#34;)
soft_limit_warned = True
if status and not chat_history: status.update(f&#34;[ai_status]Engineer: Analyzing mission... (step {iteration})&#34;)
if status and not chat_history:
status_text = f&#34;[ai_status]Engineer: Analyzing mission... (step {iteration})&#34;
if iteration &gt;= self.soft_limit_iterations:
status_text += &#34; [warning]⚠ Taking longer than expected (Ctrl+C to interrupt)[/warning]&#34;
status.update(status_text)
try:
safe_messages = self._sanitize_messages(messages)
@@ -1326,17 +1330,23 @@ class ai:
# Notificación en tiempo real de la tarea técnica (Only if not in Architect loop)
if status and not chat_history:
if fn == &#34;list_nodes&#34;: status.update(f&#34;[ai_status]Engineer: [SEARCH] {args.get(&#39;filter_pattern&#39;,&#39;.*&#39;)}&#34;)
s_text = &#34;&#34;
if fn == &#34;list_nodes&#34;: s_text = f&#34;[ai_status]Engineer: [SEARCH] {args.get(&#39;filter_pattern&#39;,&#39;.*&#39;)}&#34;
elif fn == &#34;run_commands&#34;:
cmds = args.get(&#39;commands&#39;, [])
cmd_str = cmds[0] if cmds else &#34;&#34;
status.update(f&#34;[ai_status]Engineer: [CMD] {cmd_str}&#34;)
elif fn == &#34;get_node_info&#34;: status.update(f&#34;[ai_status]Engineer: [INSPECT] {args.get(&#39;node_name&#39;,&#39;&#39;)}&#34;)
s_text = f&#34;[ai_status]Engineer: [CMD] {cmd_str}&#34;
elif fn == &#34;get_node_info&#34;: s_text = f&#34;[ai_status]Engineer: [INSPECT] {args.get(&#39;node_name&#39;,&#39;&#39;)}&#34;
elif fn.startswith(&#34;mcp_&#34;):
server = fn.split(&#34;__&#34;)[0].replace(&#34;mcp_&#34;, &#34;&#34;)
tool = fn.split(&#34;__&#34;)[1] if &#34;__&#34; in fn else fn
status.update(f&#34;[ai_status]Engineer: [MCP:{server}] {tool}&#34;)
elif fn in self.tool_status_formatters: status.update(self.tool_status_formatters[fn](args))
s_text = f&#34;[ai_status]Engineer: [MCP:{server}] {tool}&#34;
elif fn in self.tool_status_formatters: s_text = self.tool_status_formatters[fn](args)
if s_text:
if iteration &gt;= self.soft_limit_iterations:
s_text += &#34; [warning]⚠ Taking longer than expected (Ctrl+C to interrupt)[/warning]&#34;
status.update(s_text)
if debug:
self._print_debug_observation(f&#34;Decision: {fn}&#34;, args, status=status)
@@ -1406,6 +1416,8 @@ class ai:
{&#34;type&#34;: &#34;function&#34;, &#34;function&#34;: {&#34;name&#34;: &#34;return_to_engineer&#34;, &#34;description&#34;: &#34;Return control to the Engineer. Use this when your strategic analysis is complete and the Engineer should handle the rest of the conversation.&#34;, &#34;parameters&#34;: {&#34;type&#34;: &#34;object&#34;, &#34;properties&#34;: {&#34;summary&#34;: {&#34;type&#34;: &#34;string&#34;, &#34;description&#34;: &#34;Brief summary of your analysis to hand over to the Engineer.&#34;}}, &#34;required&#34;: [&#34;summary&#34;]}}},
{&#34;type&#34;: &#34;function&#34;, &#34;function&#34;: {&#34;name&#34;: &#34;manage_memory_tool&#34;, &#34;description&#34;: &#34;Saves information to long-term memory. MANDATORY: Only use this if the user explicitly asks to remember or save something.&#34;, &#34;parameters&#34;: {&#34;type&#34;: &#34;object&#34;, &#34;properties&#34;: {&#34;content&#34;: {&#34;type&#34;: &#34;string&#34;}, &#34;action&#34;: {&#34;type&#34;: &#34;string&#34;, &#34;enum&#34;: [&#34;append&#34;, &#34;replace&#34;]}}, &#34;required&#34;: [&#34;content&#34;]}}}
]
if getattr(self, &#34;one_shot&#34;, False):
base_tools = [t for t in base_tools if t[&#34;function&#34;][&#34;name&#34;] not in (&#34;delegate_to_engineer&#34;, &#34;return_to_engineer&#34;)]
all_tools = base_tools + self.external_architect_tools
seen_names = set()
@@ -1541,11 +1553,18 @@ class ai:
@MethodHook
def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=False, stream=True, session_id=None, chunk_callback=None):
soft_limit_warned = False
is_engineer_keyless = &#34;vertex&#34; in self.engineer_model.lower() or &#34;ollama&#34; in self.engineer_model.lower() or &#34;local&#34; in self.engineer_model.lower()
if not self.engineer_key and not self.engineer_auth and not is_engineer_keyless:
raise ValueError(&#34;Engineer API key or authentication not configured. Use &#39;connpy config --engineer-auth &lt;auth&gt;&#39; to set it.&#34;)
def update_status(text):
if not status:
return
if iteration &gt;= self.soft_limit_iterations:
warning_suffix = &#34; [warning]⚠ Taking longer than expected (Ctrl+C to interrupt)[/warning]&#34;
if warning_suffix not in text:
text += warning_suffix
status.update(text)
if chat_history is None: chat_history = []
@@ -1636,18 +1655,14 @@ class ai:
if self.interrupted:
raise KeyboardInterrupt
# Soft limit warning
if iteration == self.soft_limit_iterations and not soft_limit_warned:
self.console.print(f&#34;[warning]⚠ Agent has performed {iteration} steps. This is taking longer than expected.[/warning]&#34;)
self.console.print(f&#34;[warning] You can press Ctrl+C to interrupt and get a summary of progress.[/warning]&#34;)
soft_limit_warned = True
# Soft limit warning - handled inline within update_status
label = &#34;[architect][bold]Architect[/bold][/architect]&#34; if current_brain == &#34;architect&#34; else &#34;[engineer][bold]Engineer[/bold][/engineer]&#34;
if status:
# Notify responder identity for web/remote clients
if getattr(status, &#34;is_web&#34;, False) or getattr(status, &#34;is_remote&#34;, False):
status.update(f&#34;__RESPONDER__:{current_brain}&#34;)
status.update(f&#34;{label} is thinking... (step {iteration})&#34;)
update_status(f&#34;{label} is thinking... (step {iteration})&#34;)
streamed_response = False
try:
@@ -1662,7 +1677,7 @@ class ai:
response = completion(model=model, messages=safe_messages, tools=tools, num_retries=3, **current_auth)
except Exception as e:
if current_brain == &#34;architect&#34;:
if status: status.update(&#34;[unavailable]Architect unavailable! Falling back to Engineer...&#34;)
if status: update_status(&#34;[unavailable]Architect unavailable! Falling back to Engineer...&#34;)
# Preserve context when falling back - use clean_input directly
current_brain = &#34;engineer&#34;
model = self.engineer_model
@@ -1719,8 +1734,8 @@ class ai:
continue
if status:
if fn == &#34;delegate_to_engineer&#34;: status.update(f&#34;[architect]Architect: [DELEGATING MISSION] {args.get(&#39;task&#39;,&#39;&#39;)[:40]}...&#34;)
elif fn == &#34;manage_memory_tool&#34;: status.update(f&#34;[architect]Architect: [UPDATING MEMORY]&#34;)
if fn == &#34;delegate_to_engineer&#34;: update_status(f&#34;[architect]Architect: [DELEGATING MISSION] {args.get(&#39;task&#39;,&#39;&#39;)[:40]}...&#34;)
elif fn == &#34;manage_memory_tool&#34;: update_status(f&#34;[architect]Architect: [UPDATING MEMORY]&#34;)
if debug:
self._print_debug_observation(f&#34;Decision: {fn}&#34;, args, status=status)
@@ -1729,7 +1744,7 @@ class ai:
obs, eng_usage = self._engineer_loop(args[&#34;task&#34;], status=status, debug=debug, chat_history=messages[:-1])
usage[&#34;input&#34;] += eng_usage[&#34;input&#34;]; usage[&#34;output&#34;] += eng_usage[&#34;output&#34;]; usage[&#34;total&#34;] += eng_usage[&#34;total&#34;]
elif fn == &#34;consult_architect&#34;:
if status: status.update(&#34;[architect]Engineer consulting Architect...&#34;)
if status: update_status(&#34;[architect]Engineer consulting Architect...&#34;)
try:
# Consultation only - Engineer stays in control
claude_resp = completion(
@@ -1751,11 +1766,11 @@ class ai:
try: status.start()
except: pass
except Exception as e:
if status: status.update(&#34;[unavailable]Architect unavailable! Engineer continuing alone...&#34;)
if status: update_status(&#34;[unavailable]Architect unavailable! Engineer continuing alone...&#34;)
obs = f&#34;Architect unavailable ({str(e)}). Proceeding with your best technical judgment.&#34;
elif fn == &#34;escalate_to_architect&#34;:
if status: status.update(&#34;[architect]Transferring control to Architect...&#34;)
if status: update_status(&#34;[architect]Transferring control to Architect...&#34;)
# Full escalation - Architect takes over
current_brain = &#34;architect&#34;
model = self.architect_model
@@ -1777,7 +1792,7 @@ class ai:
except: pass
elif fn == &#34;return_to_engineer&#34;:
if status: status.update(&#34;[engineer]Transferring control back to Engineer...&#34;)
if status: update_status(&#34;[engineer]Transferring control back to Engineer...&#34;)
# Architect returns control to Engineer
current_brain = &#34;engineer&#34;
model = self.engineer_model
@@ -1830,7 +1845,7 @@ class ai:
messages.append(resp_msg.model_dump(exclude_none=True))
except Exception as e:
if status:
status.update(f&#34;[error]Error fetching summary: {e}[/error]&#34;)
update_status(f&#34;[error]Error fetching summary: {e}[/error]&#34;)
printer.warning(f&#34;Failed to fetch final summary from LLM: {e}&#34;)
except KeyboardInterrupt:
if status: status.update(&#34;[error]Interrupted! Closing pending tasks...&#34;)
@@ -2167,10 +2182,13 @@ Node: {node_name}&#34;&#34;&#34;
<pre><code class="python">@property
def architect_system_prompt(self):
&#34;&#34;&#34;Build architect system prompt with plugin extensions.&#34;&#34;&#34;
prompt = self._architect_base_prompt
if getattr(self, &#34;one_shot&#34;, False):
prompt += &#34;\n\nCRITICAL 1-SHOT DIAGNOSTICS DIRECTIVE:\nYou are running in a 1-shot offline diagnostics mode. There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately and directly to the user. Do not suggest or attempt to delegate/return control to the engineer.&#34;
if self.architect_prompt_extensions:
extensions = &#34;\n&#34;.join(self.architect_prompt_extensions)
return self._architect_base_prompt + f&#34;\n\nPlugin Capabilities:\n{extensions}&#34;
return self._architect_base_prompt</code></pre>
return prompt + f&#34;\n\nPlugin Capabilities:\n{extensions}&#34;
return prompt</code></pre>
</details>
<div class="desc"><p>Build architect system prompt with plugin extensions.</p></div>
</dd>
@@ -2488,11 +2506,18 @@ Node: {node_name}&#34;&#34;&#34;
</summary>
<pre><code class="python">@MethodHook
def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=False, stream=True, session_id=None, chunk_callback=None):
soft_limit_warned = False
is_engineer_keyless = &#34;vertex&#34; in self.engineer_model.lower() or &#34;ollama&#34; in self.engineer_model.lower() or &#34;local&#34; in self.engineer_model.lower()
if not self.engineer_key and not self.engineer_auth and not is_engineer_keyless:
raise ValueError(&#34;Engineer API key or authentication not configured. Use &#39;connpy config --engineer-auth &lt;auth&gt;&#39; to set it.&#34;)
def update_status(text):
if not status:
return
if iteration &gt;= self.soft_limit_iterations:
warning_suffix = &#34; [warning]⚠ Taking longer than expected (Ctrl+C to interrupt)[/warning]&#34;
if warning_suffix not in text:
text += warning_suffix
status.update(text)
if chat_history is None: chat_history = []
@@ -2583,18 +2608,14 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
if self.interrupted:
raise KeyboardInterrupt
# Soft limit warning
if iteration == self.soft_limit_iterations and not soft_limit_warned:
self.console.print(f&#34;[warning]⚠ Agent has performed {iteration} steps. This is taking longer than expected.[/warning]&#34;)
self.console.print(f&#34;[warning] You can press Ctrl+C to interrupt and get a summary of progress.[/warning]&#34;)
soft_limit_warned = True
# Soft limit warning - handled inline within update_status
label = &#34;[architect][bold]Architect[/bold][/architect]&#34; if current_brain == &#34;architect&#34; else &#34;[engineer][bold]Engineer[/bold][/engineer]&#34;
if status:
# Notify responder identity for web/remote clients
if getattr(status, &#34;is_web&#34;, False) or getattr(status, &#34;is_remote&#34;, False):
status.update(f&#34;__RESPONDER__:{current_brain}&#34;)
status.update(f&#34;{label} is thinking... (step {iteration})&#34;)
update_status(f&#34;{label} is thinking... (step {iteration})&#34;)
streamed_response = False
try:
@@ -2609,7 +2630,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
response = completion(model=model, messages=safe_messages, tools=tools, num_retries=3, **current_auth)
except Exception as e:
if current_brain == &#34;architect&#34;:
if status: status.update(&#34;[unavailable]Architect unavailable! Falling back to Engineer...&#34;)
if status: update_status(&#34;[unavailable]Architect unavailable! Falling back to Engineer...&#34;)
# Preserve context when falling back - use clean_input directly
current_brain = &#34;engineer&#34;
model = self.engineer_model
@@ -2666,8 +2687,8 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
continue
if status:
if fn == &#34;delegate_to_engineer&#34;: status.update(f&#34;[architect]Architect: [DELEGATING MISSION] {args.get(&#39;task&#39;,&#39;&#39;)[:40]}...&#34;)
elif fn == &#34;manage_memory_tool&#34;: status.update(f&#34;[architect]Architect: [UPDATING MEMORY]&#34;)
if fn == &#34;delegate_to_engineer&#34;: update_status(f&#34;[architect]Architect: [DELEGATING MISSION] {args.get(&#39;task&#39;,&#39;&#39;)[:40]}...&#34;)
elif fn == &#34;manage_memory_tool&#34;: update_status(f&#34;[architect]Architect: [UPDATING MEMORY]&#34;)
if debug:
self._print_debug_observation(f&#34;Decision: {fn}&#34;, args, status=status)
@@ -2676,7 +2697,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
obs, eng_usage = self._engineer_loop(args[&#34;task&#34;], status=status, debug=debug, chat_history=messages[:-1])
usage[&#34;input&#34;] += eng_usage[&#34;input&#34;]; usage[&#34;output&#34;] += eng_usage[&#34;output&#34;]; usage[&#34;total&#34;] += eng_usage[&#34;total&#34;]
elif fn == &#34;consult_architect&#34;:
if status: status.update(&#34;[architect]Engineer consulting Architect...&#34;)
if status: update_status(&#34;[architect]Engineer consulting Architect...&#34;)
try:
# Consultation only - Engineer stays in control
claude_resp = completion(
@@ -2698,11 +2719,11 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
try: status.start()
except: pass
except Exception as e:
if status: status.update(&#34;[unavailable]Architect unavailable! Engineer continuing alone...&#34;)
if status: update_status(&#34;[unavailable]Architect unavailable! Engineer continuing alone...&#34;)
obs = f&#34;Architect unavailable ({str(e)}). Proceeding with your best technical judgment.&#34;
elif fn == &#34;escalate_to_architect&#34;:
if status: status.update(&#34;[architect]Transferring control to Architect...&#34;)
if status: update_status(&#34;[architect]Transferring control to Architect...&#34;)
# Full escalation - Architect takes over
current_brain = &#34;architect&#34;
model = self.architect_model
@@ -2724,7 +2745,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
except: pass
elif fn == &#34;return_to_engineer&#34;:
if status: status.update(&#34;[engineer]Transferring control back to Engineer...&#34;)
if status: update_status(&#34;[engineer]Transferring control back to Engineer...&#34;)
# Architect returns control to Engineer
current_brain = &#34;engineer&#34;
model = self.engineer_model
@@ -2777,7 +2798,7 @@ def ask(self, user_input, dryrun=False, chat_history=None, status=None, debug=Fa
messages.append(resp_msg.model_dump(exclude_none=True))
except Exception as e:
if status:
status.update(f&#34;[error]Error fetching summary: {e}[/error]&#34;)
update_status(f&#34;[error]Error fetching summary: {e}[/error]&#34;)
printer.warning(f&#34;Failed to fetch final summary from LLM: {e}&#34;)
except KeyboardInterrupt:
if status: status.update(&#34;[error]Interrupted! Closing pending tasks...&#34;)
@@ -6384,6 +6405,7 @@ def test(self, commands, expected, vars = None,*, folder = None, prompt = None,
<li><a href="#ai-programmatic-use">AI Programmatic Use</a></li>
</ul>
</li>
<li><a href="#license">📜 License</a></li>
</ul>
</li>
</ul>
+102 -1
View File
@@ -369,7 +369,41 @@ el.replaceWith(d);
&#34;&#34;&#34;Load a session&#39;s raw data by ID.&#34;&#34;&#34;
from connpy.ai import ai
agent = ai(self.config)
return agent.load_session_data(session_id)</code></pre>
return agent.load_session_data(session_id)
def build_playbook_chat(self, user_input: str, chat_history: list = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Interact with the specialized Playbook Builder Agent.&#34;&#34;&#34;
from connpy.ai import PlaybookBuilderAgent
agent = PlaybookBuilderAgent(self.config)
return agent.ask(user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)
def analyze_execution_results(self, results: dict, query: str = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Analyze actual command execution results using Network Architect 1-shot.&#34;&#34;&#34;
import json
results_str = json.dumps(results, indent=2)
prompt = f&#34;@architect: Please analyze the following actual execution results. Diagnose any issues, highlight successful actions, and suggest strategic remediation steps if needed.&#34;
if query:
prompt += f&#34;\nSpecific user request: {query}&#34;
prompt += f&#34;\n\nResults Data:\n{results_str}&#34;
prompt += &#34;\n\nCRITICAL DIRECTIVE: You are running in a strictly 1-shot offline diagnostics mode (--analyze). There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately. DO NOT suggest, mention, or attempt to delegate the session back to the engineer.&#34;
# Delegate to self.ask, setting stream=True and forwarding callback/status.
# This will invoke standard ai.ask with &#39;@architect:&#39; prefix, forcing 1-shot architect brain.
return self.ask(prompt, status=status, chunk_callback=chunk_callback, one_shot=True)
def predict_execution_results(self, target_nodes: list, commands: list, status=None, chunk_callback=None):
&#34;&#34;&#34;Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).&#34;&#34;&#34;
nodes_str = &#34;, &#34;.join(target_nodes)
commands_str = &#34;\n&#34;.join(f&#34;- {cmd}&#34; for cmd in commands)
prompt = f&#34;@engineer: Act as a Preflight Simulation Agent. Simulate and predict the expected outputs and behaviors of the following commands on the target nodes. Alert about potential safety or configuration risks based on node profiles.&#34;
prompt += f&#34;\n\nTarget Nodes: {nodes_str}&#34;
prompt += f&#34;\nCommands to simulate:\n{commands_str}&#34;
prompt += &#34;\n\nCRITICAL SCALABILITY DIRECTIVE: If there are many target nodes, DO NOT list predictions node-by-node. Instead, group them by Operating System, vendor, or platform, and provide a highly concise Executive Summary. Detail individual risks only for nodes that present specific anomalies or security concerns. Focus on overall impact.&#34;
# Delegate to self.ask, using the standard engineer brain but with the simulated preflight prompt.
return self.ask(prompt, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Business logic for interacting with AI agents and LLM configurations.</p>
<p>Initialize the service.</p>
@@ -402,6 +436,31 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Ask the AI copilot for terminal assistance asynchronously.</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.analyze_execution_results"><code class="name flex">
<span>def <span class="ident">analyze_execution_results</span></span>(<span>self, results: dict, query: str = None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def analyze_execution_results(self, results: dict, query: str = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Analyze actual command execution results using Network Architect 1-shot.&#34;&#34;&#34;
import json
results_str = json.dumps(results, indent=2)
prompt = f&#34;@architect: Please analyze the following actual execution results. Diagnose any issues, highlight successful actions, and suggest strategic remediation steps if needed.&#34;
if query:
prompt += f&#34;\nSpecific user request: {query}&#34;
prompt += f&#34;\n\nResults Data:\n{results_str}&#34;
prompt += &#34;\n\nCRITICAL DIRECTIVE: You are running in a strictly 1-shot offline diagnostics mode (--analyze). There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately. DO NOT suggest, mention, or attempt to delegate the session back to the engineer.&#34;
# Delegate to self.ask, setting stream=True and forwarding callback/status.
# This will invoke standard ai.ask with &#39;@architect:&#39; prefix, forcing 1-shot architect brain.
return self.ask(prompt, status=status, chunk_callback=chunk_callback, one_shot=True)</code></pre>
</details>
<div class="desc"><p>Analyze actual command execution results using Network Architect 1-shot.</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.ask"><code class="name flex">
<span>def <span class="ident">ask</span></span>(<span>self,<br>input_text,<br>dryrun=False,<br>chat_history=None,<br>status=None,<br>debug=False,<br>session_id=None,<br>console=None,<br>chunk_callback=None,<br>confirm_handler=None,<br>trust=False,<br>**overrides)</span>
</code></dt>
@@ -559,6 +618,22 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Identifies command blocks in the terminal history.</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.build_playbook_chat"><code class="name flex">
<span>def <span class="ident">build_playbook_chat</span></span>(<span>self, user_input: str, chat_history: list = None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def build_playbook_chat(self, user_input: str, chat_history: list = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Interact with the specialized Playbook Builder Agent.&#34;&#34;&#34;
from connpy.ai import PlaybookBuilderAgent
agent = PlaybookBuilderAgent(self.config)
return agent.ask(user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Interact with the specialized Playbook Builder Agent.</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.configure_mcp"><code class="name flex">
<span>def <span class="ident">configure_mcp</span></span>(<span>self, name, url=None, enabled=None, auto_load_on_os=None, remove=False)</span>
</code></dt>
@@ -715,6 +790,29 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Load a session's raw data by ID.</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.predict_execution_results"><code class="name flex">
<span>def <span class="ident">predict_execution_results</span></span>(<span>self, target_nodes: list, commands: list, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def predict_execution_results(self, target_nodes: list, commands: list, status=None, chunk_callback=None):
&#34;&#34;&#34;Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).&#34;&#34;&#34;
nodes_str = &#34;, &#34;.join(target_nodes)
commands_str = &#34;\n&#34;.join(f&#34;- {cmd}&#34; for cmd in commands)
prompt = f&#34;@engineer: Act as a Preflight Simulation Agent. Simulate and predict the expected outputs and behaviors of the following commands on the target nodes. Alert about potential safety or configuration risks based on node profiles.&#34;
prompt += f&#34;\n\nTarget Nodes: {nodes_str}&#34;
prompt += f&#34;\nCommands to simulate:\n{commands_str}&#34;
prompt += &#34;\n\nCRITICAL SCALABILITY DIRECTIVE: If there are many target nodes, DO NOT list predictions node-by-node. Instead, group them by Operating System, vendor, or platform, and provide a highly concise Executive Summary. Detail individual risks only for nodes that present specific anomalies or security concerns. Focus on overall impact.&#34;
# Delegate to self.ask, using the standard engineer brain but with the simulated preflight prompt.
return self.ask(prompt, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).</p></div>
</dd>
<dt id="connpy.services.ai_service.AIService.process_copilot_input"><code class="name flex">
<span>def <span class="ident">process_copilot_input</span></span>(<span>self, input_text: str, session_state: dict) > dict</span>
</code></dt>
@@ -813,9 +911,11 @@ el.replaceWith(d);
<h4><code><a title="connpy.services.ai_service.AIService" href="#connpy.services.ai_service.AIService">AIService</a></code></h4>
<ul class="">
<li><code><a title="connpy.services.ai_service.AIService.aask_copilot" href="#connpy.services.ai_service.AIService.aask_copilot">aask_copilot</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.analyze_execution_results" href="#connpy.services.ai_service.AIService.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.ask" href="#connpy.services.ai_service.AIService.ask">ask</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.ask_copilot" href="#connpy.services.ai_service.AIService.ask_copilot">ask_copilot</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.build_context_blocks" href="#connpy.services.ai_service.AIService.build_context_blocks">build_context_blocks</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.build_playbook_chat" href="#connpy.services.ai_service.AIService.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.configure_mcp" href="#connpy.services.ai_service.AIService.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.configure_provider" href="#connpy.services.ai_service.AIService.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.confirm" href="#connpy.services.ai_service.AIService.confirm">confirm</a></code></li>
@@ -823,6 +923,7 @@ el.replaceWith(d);
<li><code><a title="connpy.services.ai_service.AIService.list_mcp_servers" href="#connpy.services.ai_service.AIService.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.list_sessions" href="#connpy.services.ai_service.AIService.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.load_session_data" href="#connpy.services.ai_service.AIService.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.predict_execution_results" href="#connpy.services.ai_service.AIService.predict_execution_results">predict_execution_results</a></code></li>
<li><code><a title="connpy.services.ai_service.AIService.process_copilot_input" href="#connpy.services.ai_service.AIService.process_copilot_input">process_copilot_input</a></code></li>
</ul>
</li>
+1 -110
View File
@@ -156,56 +156,7 @@ el.replaceWith(d);
except Exception as e:
raise ConnpyError(f&#34;Failed to read script {script_path}: {e}&#34;)
return self.run_commands(nodes_filter, commands, parallel=parallel)
def run_yaml_playbook(self, playbook_data: str, parallel: int = 10) -&gt; Dict[str, Any]:
&#34;&#34;&#34;Run a structured Connpy YAML automation playbook (from path or content).&#34;&#34;&#34;
playbook = None
if playbook_data.startswith(&#34;---YAML---\n&#34;):
try:
content = playbook_data[len(&#34;---YAML---\n&#34;):]
playbook = yaml.load(content, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to parse YAML content: {e}&#34;)
else:
if not os.path.exists(playbook_data):
raise ConnpyError(f&#34;Playbook file not found: {playbook_data}&#34;)
try:
with open(playbook_data, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to load playbook {playbook_data}: {e}&#34;)
# Basic validation
if not isinstance(playbook, dict) or &#34;nodes&#34; not in playbook or &#34;commands&#34; not in playbook:
raise ConnpyError(&#34;Invalid playbook format: missing &#39;nodes&#39; or &#39;commands&#39; keys.&#34;)
action = playbook.get(&#34;action&#34;, &#34;run&#34;)
options = playbook.get(&#34;options&#34;, {})
# Extract all fields similar to RunHandler.cli_run
exec_args = {
&#34;nodes_filter&#34;: playbook[&#34;nodes&#34;],
&#34;commands&#34;: playbook[&#34;commands&#34;],
&#34;variables&#34;: playbook.get(&#34;variables&#34;),
&#34;parallel&#34;: options.get(&#34;parallel&#34;, parallel),
&#34;timeout&#34;: playbook.get(&#34;timeout&#34;, options.get(&#34;timeout&#34;, 20)),
&#34;prompt&#34;: options.get(&#34;prompt&#34;),
&#34;name&#34;: playbook.get(&#34;name&#34;, &#34;Task&#34;)
}
# Map &#39;output&#39; field to folder path if it&#39;s not stdout/null
output_cfg = playbook.get(&#34;output&#34;)
if output_cfg not in [None, &#34;stdout&#34;]:
exec_args[&#34;folder&#34;] = output_cfg
if action == &#34;run&#34;:
return self.run_commands(**exec_args)
elif action == &#34;test&#34;:
exec_args[&#34;expected&#34;] = playbook.get(&#34;expected&#34;, [])
return self.test_commands(**exec_args)
else:
raise ConnpyError(f&#34;Unsupported playbook action: {action}&#34;)</code></pre>
return self.run_commands(nodes_filter, commands, parallel=parallel)</code></pre>
</details>
<div class="desc"><p>Business logic for executing commands on nodes and running automation scripts.</p>
<p>Initialize the service.</p>
@@ -300,65 +251,6 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Execute commands on a set of nodes.</p></div>
</dd>
<dt id="connpy.services.execution_service.ExecutionService.run_yaml_playbook"><code class="name flex">
<span>def <span class="ident">run_yaml_playbook</span></span>(<span>self, playbook_data: str, parallel: int = 10) > Dict[str, Any]</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def run_yaml_playbook(self, playbook_data: str, parallel: int = 10) -&gt; Dict[str, Any]:
&#34;&#34;&#34;Run a structured Connpy YAML automation playbook (from path or content).&#34;&#34;&#34;
playbook = None
if playbook_data.startswith(&#34;---YAML---\n&#34;):
try:
content = playbook_data[len(&#34;---YAML---\n&#34;):]
playbook = yaml.load(content, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to parse YAML content: {e}&#34;)
else:
if not os.path.exists(playbook_data):
raise ConnpyError(f&#34;Playbook file not found: {playbook_data}&#34;)
try:
with open(playbook_data, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to load playbook {playbook_data}: {e}&#34;)
# Basic validation
if not isinstance(playbook, dict) or &#34;nodes&#34; not in playbook or &#34;commands&#34; not in playbook:
raise ConnpyError(&#34;Invalid playbook format: missing &#39;nodes&#39; or &#39;commands&#39; keys.&#34;)
action = playbook.get(&#34;action&#34;, &#34;run&#34;)
options = playbook.get(&#34;options&#34;, {})
# Extract all fields similar to RunHandler.cli_run
exec_args = {
&#34;nodes_filter&#34;: playbook[&#34;nodes&#34;],
&#34;commands&#34;: playbook[&#34;commands&#34;],
&#34;variables&#34;: playbook.get(&#34;variables&#34;),
&#34;parallel&#34;: options.get(&#34;parallel&#34;, parallel),
&#34;timeout&#34;: playbook.get(&#34;timeout&#34;, options.get(&#34;timeout&#34;, 20)),
&#34;prompt&#34;: options.get(&#34;prompt&#34;),
&#34;name&#34;: playbook.get(&#34;name&#34;, &#34;Task&#34;)
}
# Map &#39;output&#39; field to folder path if it&#39;s not stdout/null
output_cfg = playbook.get(&#34;output&#34;)
if output_cfg not in [None, &#34;stdout&#34;]:
exec_args[&#34;folder&#34;] = output_cfg
if action == &#34;run&#34;:
return self.run_commands(**exec_args)
elif action == &#34;test&#34;:
exec_args[&#34;expected&#34;] = playbook.get(&#34;expected&#34;, [])
return self.test_commands(**exec_args)
else:
raise ConnpyError(f&#34;Unsupported playbook action: {action}&#34;)</code></pre>
</details>
<div class="desc"><p>Run a structured Connpy YAML automation playbook (from path or content).</p></div>
</dd>
<dt id="connpy.services.execution_service.ExecutionService.test_commands"><code class="name flex">
<span>def <span class="ident">test_commands</span></span>(<span>self,<br>nodes_filter: str,<br>commands: List[str],<br>expected: List[str],<br>variables: Dict[str, Any] | None = None,<br>parallel: int = 10,<br>timeout: int = 20,<br>folder: str | None = None,<br>prompt: str | None = None,<br>on_node_complete: Callable | None = None,<br>logger: Callable | None = None,<br>name: str | None = None) > Dict[str, Dict[str, bool]]</span>
</code></dt>
@@ -439,7 +331,6 @@ el.replaceWith(d);
<ul class="">
<li><code><a title="connpy.services.execution_service.ExecutionService.run_cli_script" href="#connpy.services.execution_service.ExecutionService.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.services.execution_service.ExecutionService.run_commands" href="#connpy.services.execution_service.ExecutionService.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.services.execution_service.ExecutionService.run_yaml_playbook" href="#connpy.services.execution_service.ExecutionService.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.services.execution_service.ExecutionService.test_commands" href="#connpy.services.execution_service.ExecutionService.test_commands">test_commands</a></code></li>
</ul>
</li>
+103 -111
View File
@@ -428,7 +428,41 @@ el.replaceWith(d);
&#34;&#34;&#34;Load a session&#39;s raw data by ID.&#34;&#34;&#34;
from connpy.ai import ai
agent = ai(self.config)
return agent.load_session_data(session_id)</code></pre>
return agent.load_session_data(session_id)
def build_playbook_chat(self, user_input: str, chat_history: list = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Interact with the specialized Playbook Builder Agent.&#34;&#34;&#34;
from connpy.ai import PlaybookBuilderAgent
agent = PlaybookBuilderAgent(self.config)
return agent.ask(user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)
def analyze_execution_results(self, results: dict, query: str = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Analyze actual command execution results using Network Architect 1-shot.&#34;&#34;&#34;
import json
results_str = json.dumps(results, indent=2)
prompt = f&#34;@architect: Please analyze the following actual execution results. Diagnose any issues, highlight successful actions, and suggest strategic remediation steps if needed.&#34;
if query:
prompt += f&#34;\nSpecific user request: {query}&#34;
prompt += f&#34;\n\nResults Data:\n{results_str}&#34;
prompt += &#34;\n\nCRITICAL DIRECTIVE: You are running in a strictly 1-shot offline diagnostics mode (--analyze). There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately. DO NOT suggest, mention, or attempt to delegate the session back to the engineer.&#34;
# Delegate to self.ask, setting stream=True and forwarding callback/status.
# This will invoke standard ai.ask with &#39;@architect:&#39; prefix, forcing 1-shot architect brain.
return self.ask(prompt, status=status, chunk_callback=chunk_callback, one_shot=True)
def predict_execution_results(self, target_nodes: list, commands: list, status=None, chunk_callback=None):
&#34;&#34;&#34;Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).&#34;&#34;&#34;
nodes_str = &#34;, &#34;.join(target_nodes)
commands_str = &#34;\n&#34;.join(f&#34;- {cmd}&#34; for cmd in commands)
prompt = f&#34;@engineer: Act as a Preflight Simulation Agent. Simulate and predict the expected outputs and behaviors of the following commands on the target nodes. Alert about potential safety or configuration risks based on node profiles.&#34;
prompt += f&#34;\n\nTarget Nodes: {nodes_str}&#34;
prompt += f&#34;\nCommands to simulate:\n{commands_str}&#34;
prompt += &#34;\n\nCRITICAL SCALABILITY DIRECTIVE: If there are many target nodes, DO NOT list predictions node-by-node. Instead, group them by Operating System, vendor, or platform, and provide a highly concise Executive Summary. Detail individual risks only for nodes that present specific anomalies or security concerns. Focus on overall impact.&#34;
# Delegate to self.ask, using the standard engineer brain but with the simulated preflight prompt.
return self.ask(prompt, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Business logic for interacting with AI agents and LLM configurations.</p>
<p>Initialize the service.</p>
@@ -461,6 +495,31 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Ask the AI copilot for terminal assistance asynchronously.</p></div>
</dd>
<dt id="connpy.services.AIService.analyze_execution_results"><code class="name flex">
<span>def <span class="ident">analyze_execution_results</span></span>(<span>self, results: dict, query: str = None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def analyze_execution_results(self, results: dict, query: str = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Analyze actual command execution results using Network Architect 1-shot.&#34;&#34;&#34;
import json
results_str = json.dumps(results, indent=2)
prompt = f&#34;@architect: Please analyze the following actual execution results. Diagnose any issues, highlight successful actions, and suggest strategic remediation steps if needed.&#34;
if query:
prompt += f&#34;\nSpecific user request: {query}&#34;
prompt += f&#34;\n\nResults Data:\n{results_str}&#34;
prompt += &#34;\n\nCRITICAL DIRECTIVE: You are running in a strictly 1-shot offline diagnostics mode (--analyze). There is no active conversation loop, and you are NOT conversing with a Network Engineer. You MUST deliver your complete strategic analysis immediately. DO NOT suggest, mention, or attempt to delegate the session back to the engineer.&#34;
# Delegate to self.ask, setting stream=True and forwarding callback/status.
# This will invoke standard ai.ask with &#39;@architect:&#39; prefix, forcing 1-shot architect brain.
return self.ask(prompt, status=status, chunk_callback=chunk_callback, one_shot=True)</code></pre>
</details>
<div class="desc"><p>Analyze actual command execution results using Network Architect 1-shot.</p></div>
</dd>
<dt id="connpy.services.AIService.ask"><code class="name flex">
<span>def <span class="ident">ask</span></span>(<span>self,<br>input_text,<br>dryrun=False,<br>chat_history=None,<br>status=None,<br>debug=False,<br>session_id=None,<br>console=None,<br>chunk_callback=None,<br>confirm_handler=None,<br>trust=False,<br>**overrides)</span>
</code></dt>
@@ -618,6 +677,22 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Identifies command blocks in the terminal history.</p></div>
</dd>
<dt id="connpy.services.AIService.build_playbook_chat"><code class="name flex">
<span>def <span class="ident">build_playbook_chat</span></span>(<span>self, user_input: str, chat_history: list = None, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def build_playbook_chat(self, user_input: str, chat_history: list = None, status=None, chunk_callback=None):
&#34;&#34;&#34;Interact with the specialized Playbook Builder Agent.&#34;&#34;&#34;
from connpy.ai import PlaybookBuilderAgent
agent = PlaybookBuilderAgent(self.config)
return agent.ask(user_input, chat_history=chat_history, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Interact with the specialized Playbook Builder Agent.</p></div>
</dd>
<dt id="connpy.services.AIService.configure_mcp"><code class="name flex">
<span>def <span class="ident">configure_mcp</span></span>(<span>self, name, url=None, enabled=None, auto_load_on_os=None, remove=False)</span>
</code></dt>
@@ -774,6 +849,29 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Load a session's raw data by ID.</p></div>
</dd>
<dt id="connpy.services.AIService.predict_execution_results"><code class="name flex">
<span>def <span class="ident">predict_execution_results</span></span>(<span>self, target_nodes: list, commands: list, status=None, chunk_callback=None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def predict_execution_results(self, target_nodes: list, commands: list, status=None, chunk_callback=None):
&#34;&#34;&#34;Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).&#34;&#34;&#34;
nodes_str = &#34;, &#34;.join(target_nodes)
commands_str = &#34;\n&#34;.join(f&#34;- {cmd}&#34; for cmd in commands)
prompt = f&#34;@engineer: Act as a Preflight Simulation Agent. Simulate and predict the expected outputs and behaviors of the following commands on the target nodes. Alert about potential safety or configuration risks based on node profiles.&#34;
prompt += f&#34;\n\nTarget Nodes: {nodes_str}&#34;
prompt += f&#34;\nCommands to simulate:\n{commands_str}&#34;
prompt += &#34;\n\nCRITICAL SCALABILITY DIRECTIVE: If there are many target nodes, DO NOT list predictions node-by-node. Instead, group them by Operating System, vendor, or platform, and provide a highly concise Executive Summary. Detail individual risks only for nodes that present specific anomalies or security concerns. Focus on overall impact.&#34;
# Delegate to self.ask, using the standard engineer brain but with the simulated preflight prompt.
return self.ask(prompt, status=status, chunk_callback=chunk_callback)</code></pre>
</details>
<div class="desc"><p>Predict and simulate execution results preventively using the Preflight Simulation Agent (1-shot).</p></div>
</dd>
<dt id="connpy.services.AIService.process_copilot_input"><code class="name flex">
<span>def <span class="ident">process_copilot_input</span></span>(<span>self, input_text: str, session_state: dict) > dict</span>
</code></dt>
@@ -1255,56 +1353,7 @@ el.replaceWith(d);
except Exception as e:
raise ConnpyError(f&#34;Failed to read script {script_path}: {e}&#34;)
return self.run_commands(nodes_filter, commands, parallel=parallel)
def run_yaml_playbook(self, playbook_data: str, parallel: int = 10) -&gt; Dict[str, Any]:
&#34;&#34;&#34;Run a structured Connpy YAML automation playbook (from path or content).&#34;&#34;&#34;
playbook = None
if playbook_data.startswith(&#34;---YAML---\n&#34;):
try:
content = playbook_data[len(&#34;---YAML---\n&#34;):]
playbook = yaml.load(content, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to parse YAML content: {e}&#34;)
else:
if not os.path.exists(playbook_data):
raise ConnpyError(f&#34;Playbook file not found: {playbook_data}&#34;)
try:
with open(playbook_data, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to load playbook {playbook_data}: {e}&#34;)
# Basic validation
if not isinstance(playbook, dict) or &#34;nodes&#34; not in playbook or &#34;commands&#34; not in playbook:
raise ConnpyError(&#34;Invalid playbook format: missing &#39;nodes&#39; or &#39;commands&#39; keys.&#34;)
action = playbook.get(&#34;action&#34;, &#34;run&#34;)
options = playbook.get(&#34;options&#34;, {})
# Extract all fields similar to RunHandler.cli_run
exec_args = {
&#34;nodes_filter&#34;: playbook[&#34;nodes&#34;],
&#34;commands&#34;: playbook[&#34;commands&#34;],
&#34;variables&#34;: playbook.get(&#34;variables&#34;),
&#34;parallel&#34;: options.get(&#34;parallel&#34;, parallel),
&#34;timeout&#34;: playbook.get(&#34;timeout&#34;, options.get(&#34;timeout&#34;, 20)),
&#34;prompt&#34;: options.get(&#34;prompt&#34;),
&#34;name&#34;: playbook.get(&#34;name&#34;, &#34;Task&#34;)
}
# Map &#39;output&#39; field to folder path if it&#39;s not stdout/null
output_cfg = playbook.get(&#34;output&#34;)
if output_cfg not in [None, &#34;stdout&#34;]:
exec_args[&#34;folder&#34;] = output_cfg
if action == &#34;run&#34;:
return self.run_commands(**exec_args)
elif action == &#34;test&#34;:
exec_args[&#34;expected&#34;] = playbook.get(&#34;expected&#34;, [])
return self.test_commands(**exec_args)
else:
raise ConnpyError(f&#34;Unsupported playbook action: {action}&#34;)</code></pre>
return self.run_commands(nodes_filter, commands, parallel=parallel)</code></pre>
</details>
<div class="desc"><p>Business logic for executing commands on nodes and running automation scripts.</p>
<p>Initialize the service.</p>
@@ -1399,65 +1448,6 @@ el.replaceWith(d);
</details>
<div class="desc"><p>Execute commands on a set of nodes.</p></div>
</dd>
<dt id="connpy.services.ExecutionService.run_yaml_playbook"><code class="name flex">
<span>def <span class="ident">run_yaml_playbook</span></span>(<span>self, playbook_data: str, parallel: int = 10) > Dict[str, Any]</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def run_yaml_playbook(self, playbook_data: str, parallel: int = 10) -&gt; Dict[str, Any]:
&#34;&#34;&#34;Run a structured Connpy YAML automation playbook (from path or content).&#34;&#34;&#34;
playbook = None
if playbook_data.startswith(&#34;---YAML---\n&#34;):
try:
content = playbook_data[len(&#34;---YAML---\n&#34;):]
playbook = yaml.load(content, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to parse YAML content: {e}&#34;)
else:
if not os.path.exists(playbook_data):
raise ConnpyError(f&#34;Playbook file not found: {playbook_data}&#34;)
try:
with open(playbook_data, &#34;r&#34;) as f:
playbook = yaml.load(f, Loader=yaml.FullLoader)
except Exception as e:
raise ConnpyError(f&#34;Failed to load playbook {playbook_data}: {e}&#34;)
# Basic validation
if not isinstance(playbook, dict) or &#34;nodes&#34; not in playbook or &#34;commands&#34; not in playbook:
raise ConnpyError(&#34;Invalid playbook format: missing &#39;nodes&#39; or &#39;commands&#39; keys.&#34;)
action = playbook.get(&#34;action&#34;, &#34;run&#34;)
options = playbook.get(&#34;options&#34;, {})
# Extract all fields similar to RunHandler.cli_run
exec_args = {
&#34;nodes_filter&#34;: playbook[&#34;nodes&#34;],
&#34;commands&#34;: playbook[&#34;commands&#34;],
&#34;variables&#34;: playbook.get(&#34;variables&#34;),
&#34;parallel&#34;: options.get(&#34;parallel&#34;, parallel),
&#34;timeout&#34;: playbook.get(&#34;timeout&#34;, options.get(&#34;timeout&#34;, 20)),
&#34;prompt&#34;: options.get(&#34;prompt&#34;),
&#34;name&#34;: playbook.get(&#34;name&#34;, &#34;Task&#34;)
}
# Map &#39;output&#39; field to folder path if it&#39;s not stdout/null
output_cfg = playbook.get(&#34;output&#34;)
if output_cfg not in [None, &#34;stdout&#34;]:
exec_args[&#34;folder&#34;] = output_cfg
if action == &#34;run&#34;:
return self.run_commands(**exec_args)
elif action == &#34;test&#34;:
exec_args[&#34;expected&#34;] = playbook.get(&#34;expected&#34;, [])
return self.test_commands(**exec_args)
else:
raise ConnpyError(f&#34;Unsupported playbook action: {action}&#34;)</code></pre>
</details>
<div class="desc"><p>Run a structured Connpy YAML automation playbook (from path or content).</p></div>
</dd>
<dt id="connpy.services.ExecutionService.test_commands"><code class="name flex">
<span>def <span class="ident">test_commands</span></span>(<span>self,<br>nodes_filter: str,<br>commands: List[str],<br>expected: List[str],<br>variables: Dict[str, Any] | None = None,<br>parallel: int = 10,<br>timeout: int = 20,<br>folder: str | None = None,<br>prompt: str | None = None,<br>on_node_complete: Callable | None = None,<br>logger: Callable | None = None,<br>name: str | None = None) > Dict[str, Dict[str, bool]]</span>
</code></dt>
@@ -4006,9 +3996,11 @@ el.replaceWith(d);
<h4><code><a title="connpy.services.AIService" href="#connpy.services.AIService">AIService</a></code></h4>
<ul class="">
<li><code><a title="connpy.services.AIService.aask_copilot" href="#connpy.services.AIService.aask_copilot">aask_copilot</a></code></li>
<li><code><a title="connpy.services.AIService.analyze_execution_results" href="#connpy.services.AIService.analyze_execution_results">analyze_execution_results</a></code></li>
<li><code><a title="connpy.services.AIService.ask" href="#connpy.services.AIService.ask">ask</a></code></li>
<li><code><a title="connpy.services.AIService.ask_copilot" href="#connpy.services.AIService.ask_copilot">ask_copilot</a></code></li>
<li><code><a title="connpy.services.AIService.build_context_blocks" href="#connpy.services.AIService.build_context_blocks">build_context_blocks</a></code></li>
<li><code><a title="connpy.services.AIService.build_playbook_chat" href="#connpy.services.AIService.build_playbook_chat">build_playbook_chat</a></code></li>
<li><code><a title="connpy.services.AIService.configure_mcp" href="#connpy.services.AIService.configure_mcp">configure_mcp</a></code></li>
<li><code><a title="connpy.services.AIService.configure_provider" href="#connpy.services.AIService.configure_provider">configure_provider</a></code></li>
<li><code><a title="connpy.services.AIService.confirm" href="#connpy.services.AIService.confirm">confirm</a></code></li>
@@ -4016,6 +4008,7 @@ el.replaceWith(d);
<li><code><a title="connpy.services.AIService.list_mcp_servers" href="#connpy.services.AIService.list_mcp_servers">list_mcp_servers</a></code></li>
<li><code><a title="connpy.services.AIService.list_sessions" href="#connpy.services.AIService.list_sessions">list_sessions</a></code></li>
<li><code><a title="connpy.services.AIService.load_session_data" href="#connpy.services.AIService.load_session_data">load_session_data</a></code></li>
<li><code><a title="connpy.services.AIService.predict_execution_results" href="#connpy.services.AIService.predict_execution_results">predict_execution_results</a></code></li>
<li><code><a title="connpy.services.AIService.process_copilot_input" href="#connpy.services.AIService.process_copilot_input">process_copilot_input</a></code></li>
</ul>
</li>
@@ -4041,7 +4034,6 @@ el.replaceWith(d);
<ul class="">
<li><code><a title="connpy.services.ExecutionService.run_cli_script" href="#connpy.services.ExecutionService.run_cli_script">run_cli_script</a></code></li>
<li><code><a title="connpy.services.ExecutionService.run_commands" href="#connpy.services.ExecutionService.run_commands">run_commands</a></code></li>
<li><code><a title="connpy.services.ExecutionService.run_yaml_playbook" href="#connpy.services.ExecutionService.run_yaml_playbook">run_yaml_playbook</a></code></li>
<li><code><a title="connpy.services.ExecutionService.test_commands" href="#connpy.services.ExecutionService.test_commands">test_commands</a></code></li>
</ul>
</li>