Loading the notebook…
Loading the notebook…
Notebooks · Workspace, live UI and observability · 78
Run this before anything else, and re-run it after a kernel restart.
Its imports and arguments are checked against shipit-agent 3.0. The outputs were saved when it was written, for an earlier release.
# Setup — run this FIRST.
#
# Order matters. Jupyter imports whatever `shipit_agent` is installed in the
# kernel, which lags this checkout: `include_server_in_tool_names` exists in
# the repo and not in the released package, so RemoteMCPServer(...) raises
# TypeError on an argument that is genuinely there. Putting the repo ahead of
# site-packages BEFORE the first import is the whole fix — adjusting sys.path
# afterwards is too late, the wrong module is already cached.
import sys
import pathlib
repo = pathlib.Path.cwd().parent
if str(repo) not in sys.path:
sys.path.insert(0, str(repo))
import shipit_agent
print(f"shipit-agent {shipit_agent.__version__}")
print(f"from {shipit_agent.__file__}")
assert str(repo) in shipit_agent.__file__, (
"still importing the installed copy — restart the kernel and run this cell first"
)shipit-agent 1.9.8 from ~/Documents/MYWORK/ai_developer/others/shipit_agent/shipit_agent/__init__.py
End-to-end validation of the main Agent with Gemma 4 on Bedrock Mantle and the public DeepWiki MCP. This notebook prints the complete tool/event stream, measures token usage, verifies canonical output retention, and exercises failure and large-output paths.
_meta without leaking it into tool argumentsrun_failed eventsimport importlib.util
import json
import os
import sys
import time
from collections import Counter
from pathlib import Path
from shipit_agent import Agent, FunctionTool, format_event_line
from shipit_agent.llms import LLMResponse, LiteLLMChatLLM
from shipit_agent.mcp import MCPStreamableHTTPTransport, RemoteMCPServer
from shipit_agent.models import ToolCall
from shipit_agent.tools import ToolContext
print('Python:', sys.version.split()[0])
print('Workspace:', Path.cwd())Python: 3.11.9 Workspace: ~/Documents/MYWORK/ai_developer/others/shipit_agent/notebooks
The path and model are environment-overridable. No API key or credential is embedded in this notebook.
DEFAULT_PROVIDER = (
'~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK/'
'drk_cache/llm/bedrock_mantle_provider.py'
)
PROVIDER = Path(os.getenv('SHIPIT_MANTLE_PROVIDER', DEFAULT_PROVIDER))
MODEL = os.getenv('SHIPIT_AUDIT_MODEL', 'bedrock-mantle/google.gemma-4-26b-a4b')
assert PROVIDER.exists(), f'Provider not found: {PROVIDER}'
spec = importlib.util.spec_from_file_location('bedrock_mantle_provider', PROVIDER)
module = importlib.util.module_from_spec(spec)
sys.modules['bedrock_mantle_provider'] = module
spec.loader.exec_module(module)
module.ensure_registered()
def new_llm():
return LiteLLMChatLLM(model=MODEL)
print('Provider:', PROVIDER)
print('Model:', MODEL)Provider: ~/Documents/MYWORK/AFTDRK/CACHE/DRK_CACHE_BACK/drk_cache/llm/bedrock_mantle_provider.py Model: bedrock-mantle/google.gemma-4-26b-a4b
DeepWiki is public and needs no authentication. The metadata resolver demonstrates generic trace context attached through MCP _meta.
# 1. Deferred tool loading — a small core stays resident; the rest (and MCP
# tools) are listed by name and loaded on demand via tool_search.
agent = Agent.with_builtins(llm=new_llm(), deferred_tools=True, observability=False)
for i in agent.stream('What is this image and both documenr about ? read all docuem an explain in depth', images=["~/Desktop/Payment_Confirmation.png"], files=["~/Documents/MYWORK/ai_developer/others/shipit_agent/docs/deep-agents/goal-agent.md", "~/Downloads/Mrs_MEGHA_SINGH__10_08_2026_03_12_46_PM.pdf"]):
print(i)
AgentEvent(type='run_started', message='Agent run started', payload={'prompt': 'What is this image and both documenr about ? read all docuem an explain in depth'}, timestamp=1787135161.951942)
AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 24, 'iteration': 1}, timestamp=1787135161.9564722)
AgentEvent(type='text_delta', message='', payload={'chunk': 'The'}, timestamp=1787135165.0587451)
AgentEvent(type='text_delta', message='', payload={'chunk': ' provided'}, timestamp=1787135165.251631)
AgentEvent(type='text_delta', message='', payload={'chunk': ' materials'}, timestamp=1787135165.253182)
AgentEvent(type='text_delta', message='', payload={'chunk': ' consist'}, timestamp=1787135165.254824)
AgentEvent(type='text_delta', message='', payload={'chunk': ' of'}, timestamp=1787135165.255841)
AgentEvent(type='text_delta', message='', payload={'chunk': ' a'}, timestamp=1787135165.25696)
AgentEvent(type='text_delta', message='', payload={'chunk': ' payment'}, timestamp=1787135165.2585468)
AgentEvent(type='text_delta', message='', payload={'chunk': ' confirmation'}, timestamp=1787135165.260094)
AgentEvent(type='text_delta', message='', payload={'chunk': ' image'}, timestamp=1787135165.261369)
AgentEvent(type='text_delta', message='', payload={'chunk': ' and'}, timestamp=1787135165.26259)
AgentEvent(type='text_delta', message='', payload={'chunk': ' a'}, timestamp=1787135165.2643049)
AgentEvent(type='text_delta', message='', payload={'chunk': ' technical'}, timestamp=1787135165.265266)
AgentEvent(type='text_delta', message='', payload={'chunk': ' documentation'}, timestamp=1787135165.266748)
AgentEvent(type='text_delta', message='', payload={'chunk': ' file'}, timestamp=1787135165.2685719)
AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1787135165.269089)
AgentEvent(type='text_delta', message='', payload={'chunk': '\n'}, timestamp=1787135165.374882)
AgentEvent(type='text_delta', message='', payload={'chunk': '\n###'}, timestamp=1787135165.375595)
AgentEvent(type='text_delta', message='', payload={'chunk': ' '}, timestamp=1787135165.376073)
AgentEvent(type='text_delta', message='', payload={'chunk': '1'}, timestamp=1787135165.376418)
AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1787135165.376737)
AgentEvent(type='text_delta', message='', payload={'chunk': ' Image'}, timestamp=1787135165.3772428)
AgentEvent(type='text_delta', message='', payload={'chunk': ' Analysis'}, timestamp=1787135165.377929)
AgentEvent(type='text_delta', message='', payload={'chunk': ':'}, timestamp=1787135165.4539719)
AgentEvent(type='text_delta', message='', payload={'chunk': ' Payment'}, timestamp=1787135165.4555762)
AgentEvent(type='text_delta', message='', payload={'chunk': ' Confirmation'}, timestamp=1787135165.457072)
AgentEvent(type='text_delta', message='', payload={'chunk': '\nThe'}, timestamp=1787135165.457969)
AgentEvent(type='text_delta', message='', payload={'chunk': ' image'}, timestamp=1787135165.458913)
AgentEvent(type='text_delta', message='', payload={'chunk': ' is'}, timestamp=1787135165.460453)
AgentEvent(type='text_delta', message='', payload={'chunk': ' a'}, timestamp=1787135165.461486)
AgentEvent(type='text_delta', message='', payload={'chunk': ' digital'}, timestamp=1787135165.4634151)
AgentEvent(type='text_delta', message='', payload={'chunk': ' receipt'}, timestamp=1787135165.465716)
AgentEvent(type='text_delta', message='', payload={'chunk': ' for'}, timestamp=1787135165.467208)
AgentEvent(type='text_delta', message='', payload={'chunk': ' a'}, timestamp=1787135165.467718)
AgentEvent(type='text_delta', message='', payload={'chunk': ' financial'}, timestamp=1787135165.469095)
AgentEvent(type='text_delta', message='', payload={'chunk': ' transaction'}, timestamp=1787135165.46985)
AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1787135165.4706619)
AgentEvent(type='text_delta', message='', payload={'chunk': ' likely'},
… output trimmedNo pricing data for model 'bedrock-mantle/google.gemma-4-26b-a4b' — cost will be $0.00
AgentEvent(type='text_delta', message='', payload={'chunk': ' agents'}, timestamp=1787135170.253771)
AgentEvent(type='text_delta', message='', payload={'chunk': ' that'}, timestamp=1787135170.2551548)
AgentEvent(type='text_delta', message='', payload={'chunk': ' can'}, timestamp=1787135170.256351)
AgentEvent(type='text_delta', message='', payload={'chunk': ' reason'}, timestamp=1787135170.257417)
AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1787135170.2579908)
AgentEvent(type='text_delta', message='', payload={'chunk': ' use'}, timestamp=1787135170.259063)
AgentEvent(type='text_delta', message='', payload={'chunk': ' tools'}, timestamp=1787135170.2600698)
AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1787135170.2606442)
AgentEvent(type='text_delta', message='', payload={'chunk': ' remember'}, timestamp=1787135170.261837)
AgentEvent(type='text_delta', message='', payload={'chunk': ' past'}, timestamp=1787135170.262729)
AgentEvent(type='text_delta', message='', payload={'chunk': ' interactions'}, timestamp=1787135170.2644129)
AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1787135170.264871)
AgentEvent(type='text_delta', message='', payload={'chunk': ' and'}, timestamp=1787135170.265412)
AgentEvent(type='text_delta', message='', payload={'chunk': ' verify'}, timestamp=1787135170.266012)
AgentEvent(type='text_delta', message='', payload={'chunk': ' their'}, timestamp=1787135170.266562)
AgentEvent(type='text_delta', message='', payload={'chunk': ' own'}, timestamp=1787135170.2670562)
AgentEvent(type='text_delta', message='', payload={'chunk': ' work'}, timestamp=1787135170.2675788)
AgentEvent(type='text_delta', message='', payload={'chunk': ' against'}, timestamp=1787135170.2686741)
AgentEvent(type='text_delta', message='', payload={'chunk': ' defined'}, timestamp=1787135170.291453)
AgentEvent(type='text_delta', message='', payload={'chunk': ' standards'}, timestamp=1787135170.29229)
AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1787135170.293141)
AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 8451, 'completion_tokens': 595, 'total_tokens': 9046, 'cache_read_input_tokens': 0, 'cache_creation_input_tokens': 0}, 'iteration': 1}, timestamp=1787135170.336499)
AgentEvent(type='final_answer', message='Final answer generated', payload={'content': 'The provided materials consist of a payment confirmation image and a technical documentation file.\n\n### 1. Image Analysis: Payment Confirmation\nThe image is a digital receipt for a financial transaction, likely from a banking or mobile payment application.\n\n* **Transaction Details:**\n * **Payment Category:** PIT (Personal Income Tax)\n * **Payment Title/Reference:** PIT-37\n * **Amount:** 2,256.00 PLN\n * **Period:** 24R (likely referring to the tax year or a specific period)\n … (2,088 more characters)
AgentEvent(type='run_summary', message='Run finished: 1 iterations, 0 tool calls, 9,046 tokens', payload={'headline': 'Run finished: 1 iterations, 0 tool calls, 9,046 tokens', 'iterations': 1, 'tool_calls': 0, 'tool_results': 0, 'compactions': 0, 'usage': {'prompt_tokens': 8451, 'completion_tokens': 595, 'total_tokens': 9046, 'cache_read_input_tokens': 0, 'cache_creation_input_tokens': 0}, 'estimated_cost_usd': 0.0, 'gave_up': False}, timestamp=1787135170.3443131)
AgentEvent(type='run_completed', message='Agent run completed', payload={'output': 'The provided materials consist of a payment confirmation image and a technical documentation file.\n\n### 1. Image Analysis: Payment Confirmation\nThe image is a digital receipt for a financial transaction, likely from a banking or mobile payment application.\n\n* **Transaction Details:**\n * **Payment Category:** PIT (Personal Income Tax)\n * **Payment Title/Reference:** PIT-37\n * **Amount:** 2,256.00 PLN\n * **Period:** 24R (likely referri
… output trimmedwatcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'# 1. Deferred tool loading — a small core stays resident; the rest (and MCP
# tools) are listed by name and loaded on demand via tool_search.
agent1 = Agent(llm=new_llm(), deferred_tools=True, observability=False)
for i in agent1.stream('What is this image and both documenr about ? read all docuem an explain in depth'):
print(i)
AgentEvent(type='run_started', message='Agent run started', payload={'prompt': 'What is this image and both documenr about ? read all docuem an explain in depth'}, timestamp=1787135953.8273642)
AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 0, 'iteration': 1}, timestamp=1787135953.827532)No pricing data for model 'bedrock-mantle/google.gemma-4-26b-a4b' — cost will be $0.00
AgentEvent(type='text_delta', message='', payload={'chunk': 'Please'}, timestamp=1787135954.596179)
AgentEvent(type='text_delta', message='', payload={'chunk': ' provide'}, timestamp=1787135954.597155)
AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1787135954.597301)
AgentEvent(type='text_delta', message='', payload={'chunk': ' image'}, timestamp=1787135954.597409)
AgentEvent(type='text_delta', message='', payload={'chunk': ' and'}, timestamp=1787135954.597578)
AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1787135954.597685)
AgentEvent(type='text_delta', message='', payload={'chunk': ' documents'}, timestamp=1787135954.597783)
AgentEvent(type='text_delta', message='', payload={'chunk': ' you'}, timestamp=1787135954.5981598)
AgentEvent(type='text_delta', message='', payload={'chunk': ' are'}, timestamp=1787135954.774045)
AgentEvent(type='text_delta', message='', payload={'chunk': ' referring'}, timestamp=1787135954.7742789)
AgentEvent(type='text_delta', message='', payload={'chunk': ' to'}, timestamp=1787135954.774397)
AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1787135954.7746408)
AgentEvent(type='text_delta', message='', payload={'chunk': ' Once'}, timestamp=1787135954.77476)
AgentEvent(type='text_delta', message='', payload={'chunk': ' you'}, timestamp=1787135954.774872)
AgentEvent(type='text_delta', message='', payload={'chunk': ' upload'}, timestamp=1787135954.774969)
AgentEvent(type='text_delta', message='', payload={'chunk': ' them'}, timestamp=1787135954.775192)
AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1787135954.7753978)
AgentEvent(type='text_delta', message='', payload={'chunk': ' I'}, timestamp=1787135954.7755032)
AgentEvent(type='text_delta', message='', payload={'chunk': ' will'}, timestamp=1787135954.775894)
AgentEvent(type='text_delta', message='', payload={'chunk': ' analyze'}, timestamp=1787135954.77601)
AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1787135954.776293)
AgentEvent(type='text_delta', message='', payload={'chunk': ' image'}, timestamp=1787135954.7763991)
AgentEvent(type='text_delta', message='', payload={'chunk': ' and'}, timestamp=1787135954.776491)
AgentEvent(type='text_delta', message='', payload={'chunk': ' perform'}, timestamp=1787135954.776584)
AgentEvent(type='text_delta', message='', payload={'chunk': ' an'}, timestamp=1787135954.776703)
AgentEvent(type='text_delta', message='', payload={'chunk': ' in'}, timestamp=1787135954.7767959)
AgentEvent(type='text_delta', message='', payload={'chunk': '-'}, timestamp=1787135954.776884)
AgentEvent(type='text_delta', message='', payload={'chunk': 'depth'}, timestamp=1787135954.77698)
AgentEvent(type='text_delta', message='', payload={'chunk': ' reading'}, timestamp=1787135954.777092)
AgentEvent(type='text_delta', message='', payload={'chunk': ' of'}, timestamp=1787135954.777184)
AgentEvent(type='text_delta', message='', payload={'chunk': ' both'}, timestamp=1787135954.777271)
AgentEvent(type='text_delta', message='', payload={'chunk': ' documents'}, timestamp=1787135954.7778502)
AgentEvent(type='text_delta', message='', payload={'chunk': ' to'}, timestamp=1787135954.778133)
AgentEvent(type='text_delta', message='', payload={'chunk': ' explain'}, timestamp=1787135954.7789388)
AgentEvent(type='text_delta', message='', payload={'chunk': ' their'}, timestamp=1787135954.779045)
AgentEvent(type='text_delta', message='', payload={'chunk': ' contents'}, timestamp=1787135954.779773)
AgentEvent(type='text_delta', message='', payload={'chunk': ' to'}, timestamp=1787135954.780193)
AgentEvent(type='text_delta', message='', payload={'chunk': ' you'}, timestamp=1787135954.780301)
AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1787135954.7803898)
AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 340, 'completion_tokens': 38, 'total_toke
… output trimmed
# # 2. Attachments — images, PDFs, and code/markdown files on the turn.
# agent.run("What is this image and both documenr about ?", images=["~/Desktop/Payment_Confirmation.png"], files=["~/Documents/MYWORK/ai_developer/others/shipit_agent/docs/deep-agents/goal-agent.md", "~/Downloads/Mrs_MEGHA_SINGH__10_08_2026_03_12_46_PM.pdf"])watcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'AgentResult(output='The image is a payment confirmation receipt, and the document (`goal-agent.md`) is a technical README for a software component called **GoalAgent**.\n\n### Image: Payment Receipt\nThe image shows a transaction confirmation for a payment made on **May 5, 2025**.\n* **Payment Title:** PIT-37\n* **Amount:** 2,256.00 PLN\n* **Method:** BLIK (Transaction number: 88071988641)\n* **Payee:** URZĄD SKARBOWY WARSZAWA-PRAGA\n* **Status:** Approved\n\n### Document: `goal-agent.md`\nThis is a documentation file for **GoalAgent**, an autonomous AI agent designed to decompose complex goal… (1,078,152 more characters)
DEEPWIKI_URL = 'https://mcp.deepwiki.com/mcp'
def new_deepwiki(*, namespaced=False):
return RemoteMCPServer(
name='deepwiki',
transport=MCPStreamableHTTPTransport(DEEPWIKI_URL, timeout=120),
include_server_in_tool_names=namespaced,
tool_meta_resolver=lambda context, tool, arguments: {
'trace_id': context.metadata.get('trace_id', 'not-set'),
'client': 'shipit-notebook-audit',
},
)
server = new_deepwiki()
started = time.perf_counter()
try:
discovered = server.discover_tools()
print('Server:', server.server_info)
print('Protocol:', server.protocol_version)
print('Discovery seconds:', round(time.perf_counter() - started, 2))
for tool in discovered:
print('-', tool.name, json.dumps(tool.input_schema))
finally:
server.close()Server: {'name': 'DeepWiki', 'version': '2.14.3'}
Protocol: 2025-11-25
Discovery seconds: 1.28
- ask_question {"properties": {"repoName": {"anyOf": [{"type": "string"}, {"items": {"type": "string"}, "type": "array"}], "description": "GitHub repository or list of repositories (max 10) in owner/repo format."}, "question": {"description": "The question to ask about the repository.", "type": "string"}}, "required": ["repoName", "question"], "type": "object"}
- read_wiki_contents {"properties": {"repoName": {"description": "GitHub repository in owner/repo format (e.g. \"facebook/react\").", "type": "string"}}, "required": ["repoName"], "type": "object"}
- read_wiki_structure {"properties": {"repoName": {"description": "GitHub repository in owner/repo format (e.g. \"facebook/react\").", "type": "string"}}, "required": ["repoName"], "type": "object"}This isolates remote retrieval latency and response size from model overhead.
Only DeepWiki tools are exposed for this focused test. Complete results remain caller-visible; only the copy entering future model turns is subject to context budgets.
QUESTION = (
'Concisely explain retry eligibility, backoff, and streaming cleanup. '
'Name the core classes and methods and give two failure tests.'
)
def new_agent():
return Agent(
llm=new_llm(),
# Narration for the diagnostic reprint further down. Only called
# when Gemma writes no prose of its own, which for a tool call is
# always — its whole completion is the call.
decision_llm=new_llm(),
mcps=[new_deepwiki(namespaced=True)],
prompt=(
'You are a precise repository research agent. Call DeepWiki before '
'repository claims. Use declared arguments only. Ground the answer '
'in canonical tool output.'
),
name='gemma-deepwiki-notebook-audit',
metadata={'trace_id': 'notebook-live-run'},
max_iterations=6,
progress_summaries=True,
parallel_tool_execution=True,
max_tool_concurrency=3,
max_tool_output_chars=16_000,
max_tool_output_group_chars=48_000,
persist_large_tool_outputs=True,
project_root='.',
)
agent = new_agent()
TASK = (
'Use DeepWiki ask_question on openai/openai-python. ' + QUESTION +
' Call exactly one MCP tool unless it fails.'
)
selected = agent._selected_skills(TASK)
print('Selected skills:', [skill.id for skill in selected])
print('Effective max iterations:', agent._effective_max_iterations(selected))Selected skills: [] Effective max iterations: 6
The MCP body is printed once from tool_output_delta. Final text chunks print continuously as they arrive. Heavy canonical MCP metadata is intentionally not duplicated into every delta.
TASK = (
'Use DeepWiki ask_question on openai/openai-python. ' + QUESTION +
' Call exactly one MCP tool unless it fails.'
)
# `decision_llm` is what makes agent_decision read like a sentence a person
# wrote rather than a label a function composed. It is only ever called when
# the main model produced no usable prose of its own — Gemma spends its whole
# completion on the tool call, 14-16 tokens, so there is nothing to prefer and
# 'Asking question openai/openai-python.' is all a composed label can honestly
# say. Drop the argument and nothing extra is called at all.
#
# The prompt it receives is deliberately tiny: the request, the previous
# observation, and the call about to run — a few hundred characters, never the
# conversation and never the tool payloads.
narrated = Agent(
llm=new_llm(),
decision_llm=new_llm(),
mcps=[new_deepwiki(namespaced=True)],
prompt=(
'You are a precise repository research agent. Call DeepWiki before '
'repository claims. Use declared arguments only. Ground the answer '
'in canonical tool output.'
),
name='gemma-deepwiki-notebook-audit',
metadata={'trace_id': 'notebook-live-run'},
max_iterations=6,
progress_summaries=True,
parallel_tool_execution=True,
max_tool_concurrency=3,
max_tool_output_chars=16_000,
max_tool_output_group_chars=48_000,
persist_large_tool_outputs=True,
project_root='.',
)
narrated_events = []
for event in narrated.stream(TASK):
narrated_events.append(event)
if event.type == 'agent_decision':
print(f"\n\u25b8 {event.payload['summary']}")
elif event.type == 'agent_observation':
print(f" \u2190 {event.payload['summary']}")
elif event.type == 'progress_summary_failed':
print(f" ! {event.payload.get('error', '')[:140]}")
mantle rejected 'chat_template_kwargs' for google.gemma-4-26b-a4b; retrying without it (and omitting it from now on)
▸ I will ask a question to the `openai/openai-python` repository to retrieve details about retry eligibility, backoff, and streaming cleanup. xml ← Asked question openai/openai-python.
watcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'# Nothing is filtered here. The narration above is two of these events;
# every tool_group_started / tool_called / tool_completed is still present.
for event in narrated_events:
print('event', repr(event))
event AgentEvent(type='run_started', message='Agent run started', payload={'prompt': 'Use DeepWiki ask_question on openai/openai-python. Concisely explain retry eligibility, backoff, and streaming cleanup. Name the core classes and methods and give two failure tests. Call exactly one MCP tool unless it fails.'}, timestamp=1786534404.890483)
event AgentEvent(type='mcp_attached', message='MCP server attached: deepwiki', payload={'server': 'deepwiki'}, timestamp=1786534404.8915231)
event AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 3, 'iteration': 1}, timestamp=1786534404.892596)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'I'}, timestamp=1786534406.4063401)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' will'}, timestamp=1786534406.407514)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' ask'}, timestamp=1786534406.409039)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' a'}, timestamp=1786534406.410763)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' question'}, timestamp=1786534406.411911)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' to'}, timestamp=1786534406.4127)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1786534406.413923)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' '}, timestamp=1786534406.415138)
event AgentEvent(type='text_delta', message='', payload={'chunk': '`openai'}, timestamp=1786534406.416258)
event AgentEvent(type='text_delta', message='', payload={'chunk': '/'}, timestamp=1786534406.416663)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'openai'}, timestamp=1786534406.417637)
event AgentEvent(type='text_delta', message='', payload={'chunk': '-'}, timestamp=1786534406.4192052)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'python'}, timestamp=1786534406.419594)
event AgentEvent(type='text_delta', message='', payload={'chunk': '` repository'}, timestamp=1786534406.420154)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' to'}, timestamp=1786534406.420435)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' retrieve'}, timestamp=1786534406.420744)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' details'}, timestamp=1786534406.51821)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' about'}, timestamp=1786534406.518807)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' retry'}, timestamp=1786534406.5191429)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' eligibility'}, timestamp=1786534406.519744)
event AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1786534406.519983)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' back'}, timestamp=1786534406.5202408)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'off'}, timestamp=1786534406.520737)
event AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1786534406.520967)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' and'}, timestamp=1786534406.521201)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' streaming'}, timestamp=1786534406.521471)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' cleanup'}, timestamp=1786534406.522204)
event AgentEvent(type='text_delta', message='', payload={'chunk': '.'}, timestamp=1786534406.523144)
event AgentEvent(type='text_delta', message='', payload={'chunk': '\n'}, timestamp=1786534406.5241501)
event AgentEvent(type='text_delta', message='', payload={'chunk': '\nxml'}, timestamp=1786534406.525171)
event AgentEvent(type='text_delta', message='', payload={'chunk': '\n'}, timestamp=1786534406.5266979)
event AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 1297, 'completion_tokens
… output trimmedQUESTION = (
'Concisely explain retry eligibility, backoff, and streaming cleanup. '
'Name the core classes and methods and give two failure tests.'
)
server = new_deepwiki()
try:
ask = next(tool for tool in server.discover_tools() if tool.name == 'ask_question')
started = time.perf_counter()
direct = ask.run(
ToolContext(prompt=QUESTION, metadata={'trace_id': 'direct-baseline'}),
repoName='openai/openai-python',
question=QUESTION,
)
direct_seconds = time.perf_counter() - started
print({
'ok': direct.metadata.get('ok'),
'seconds': round(direct_seconds, 2),
'characters': len(direct.text),
'words': len(direct.text.split()),
})
print(direct.text)
finally:
server.close(){'ok': True, 'seconds': 19.34, 'characters': 4377, 'words': 553}
You're asking about how the OpenAI Python SDK handles retries for API requests, including which errors trigger retries, the exponential backoff mechanism, and how streaming responses are cleaned up. You also want to know the core classes and methods involved and see two failure tests.
## Retry Eligibility and Backoff
The SDK automatically retries certain errors with an exponential backoff strategy. The `BaseClient` class, specifically its `_should_retry` method, determines retry eligibility.
### Eligible Errors
The client retries requests that encounter:
* Connection errors.
* HTTP status codes:
* `408 Request Timeout`
* `409 Conflict` (lock timeouts)
* `429 Rate Limit`
* `>=500 Internal errors`
* The `x-should-retry` response header set to `"true"`.
You can configure the maximum number of retries using the `max_retries` option during client initialization or on a per-request basis.
### Exponential Backoff
The `_calculate_retry_timeout` method in `BaseClient` computes the delay before a retry. It uses an exponential backoff algorithm, starting with an `INITIAL_RETRY_DELAY` (default 0.5 seconds) and doubling it for each subsequent retry, up to a `MAX_RETRY_DELAY` (default 8 seconds). A small amount of jitter is added to prevent thundering herd problems. If the `Retry-After` header is present in the response and suggests a reasonable wait time (up to 60 seconds), that value is used instead.
The retry logic is implemented within the `request` method of `SyncAPIClient` and `AsyncAPIClient`.
## Streaming Cleanup
For streaming responses, proper resource cleanup is handled by context managers.
### Core Classes and Methods
* `Stream` (for synchronous operations) and `AsyncStream` (for asynchronous operations) are the core classes for handling streaming responses.
* Both classes implement `__enter__` and `__exit__` (or `__aenter__` and `__aexit__` for async) methods.
* The `close()` method is responsible for closing the underlying `httpx.Response` and releasing the connection. This method is automatically called when the response body is read to completion or when the context manager exits.
* In asynchronous streaming, the `__stream__` method ensures that `response.aclose()` is called in a `finally` block, guaranteeing cleanup even if the consumer doesn't read all data.
## Failure Tests
The test suite includes scenarios to verify retry behavior and connection cleanup.
1. **Retrying Timeout Errors Doesn't Leak Connections**: This test checks that if a request fails due to a `httpx.TimeoutException`, the client attempts retries, and upon final failure, no HTTP connections are left open.
* `test_retrying_timeout_errors_doesnt_leak` in `tests/test_client.py`
… 17 more linesevent_counts = Counter()
completed_payload = {}
tool_telemetry = []
events = []
in_text = False
started = time.perf_counter()
for event in agent.stream(TASK):
# print('event', event)
events.append(event)
event_counts[event.type] += 1
payload = event.payload
if event.type == 'text_delta':
if not in_text:
print('\ntext_delta stream:')
in_text = True
print(payload.get('chunk', ''), end='', flush=True)
continue
if in_text:
print('\n[end text_delta stream]')
in_text = False
if event.type == 'tool_output_delta':
chunk = str(payload.get('chunk', ''))
print('tool_output_delta', {
'tool': payload.get('tool'),
'sequence': payload.get('sequence'),
'characters': len(chunk),
'metadata': payload.get('chunk_metadata'),
})
print(chunk)
elif event.type == 'tool_completed':
telemetry = {
'tool': payload.get('tool'),
'output_chars': payload.get('output_chars'),
'model_output_chars': payload.get('model_output_chars'),
'model_output_reduced': payload.get('model_output_reduced'),
'metadata': payload.get('metadata'),
}
tool_telemetry.append(telemetry)
print('tool_completed', json.dumps(telemetry, default=str))
elif event.type == 'run_completed':
completed_payload = dict(payload)
print('run_completed', {
'usage': payload.get('usage'),
'output_chars': len(str(payload.get('output', ''))),
'cancelled': payload.get('cancelled'),
})
else:
display_line = format_event_line(event)
if display_line:
print(display_line)
else:
safe = {
key: value for key, value in payload.items()
if key not in {'output', 'content', 'chunk'}
}
print(event.type, json.dumps(safe, ensure_ascii=False, default=str))
if in_text:
print('\n[end text_delta stream]')
agent_seconds = time.perf_counter() - startedrun_started {"prompt": "Use DeepWiki ask_question on openai/openai-python. Concisely explain retry eligibility, backoff, and streaming cleanup. Name the core classes and methods and give two failure tests. Call exactly one MCP tool unless it fails."}
mcp_attached {"server": "deepwiki"}
step_started {"tool_count": 3, "iteration": 1}
usage_tick {"usage": {"prompt_tokens": 1297, "completion_tokens": 73, "total_tokens": 1370, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}, "iteration": 1}
usage_tick {"usage": {"prompt_tokens": 1540, "completion_tokens": 137, "total_tokens": 1677, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0}, "iteration": 1}
agent_decision I am about to use the `deepwiki__ask_question` tool to query the `openai/openai-python` repository for specific implementation details regarding error handling…
tool_group_started {"group_id": "tool_group_1", "iteration": 1, "tool_count": 1, "tools": [{"name": "deepwiki__ask_question", "call_id": "call_1_1"}]}
⚙ deepwiki__ask_question(repoName="openai/openai-python", question="Explain retry eligibility, backoff, …) …
tool_output_started {"tool": "deepwiki__ask_question", "call_id": "call_1_1", "group_id": "tool_group_1", "attempt": 1, "buffered": false, "iteration": 1}
tool_output_delta {'tool': 'deepwiki__ask_question', 'sequence': 1, 'characters': 3879, 'metadata': {'server': 'deepwiki', 'ok': True, 'is_error': False, 'output_schema': {'properties': {'result': {'type': 'string'}}, 'required': ['result'], 'type': 'object', 'x-fastmcp-wrap-result': True}, 'annotations': {}, 'remote_name': 'ask_question', 'title': '', 'execution': {}}}
This library handles retry eligibility, exponential backoff, and streaming cleanup primarily within the `BaseClient` and `AsyncAPIClient` classes. Streaming cleanup is specifically managed by the `AsyncStream` class.
## Retry Eligibility and Backoff
The core logic for determining retry eligibility and calculating backoff is found in the `BaseClient` class.
### Core Classes and Methods
* **`BaseClient`**: This class contains the methods responsible for retry logic.
* **`_should_retry(self, response: httpx.Response) -> bool`**: This method determines if a request should be retried based on the HTTP response. It checks for specific HTTP status codes and a custom `x-should-retry` header.
* **Retry Conditions**:
* `x-should-retry: true` header.
* HTTP status code `408` (Request Timeout).
* HTTP status code `409` (Conflict, often used for lock timeouts).
* HTTP status code `429` (Too Many Requests).
* Any HTTP status code `>= 500` (Internal Server Error, Bad Gateway, Service Unavailable, Gateway Timeout).
* **`_calculate_retry_timeout(self, remaining_retries: int, options: FinalRequestOptions, response_headers: Optional[httpx.Headers] = None) -> float`**: This method calculates the duration to wait before the next retry. It incorporates:
* **`Retry-After` header**: If present and within a reasonable range (0 to 60 seconds), it uses this value.
* **Exponential Backoff**: Otherwise, it applies an exponential backoff strategy, with jitter, capped at a maximum delay.
* **`_sleep_for_retry(...)`**: This method pauses execution for the calculated timeout duration.
* **`SyncAPIClient`** and **`AsyncAPIClient`**: These classes implement the retry loop within their `request` methods. They iterate up to `max_retries + 1` times, calling `_should_retry` and `_sleep_for_retry` as needed.
### Failure Test Examples
1. **HTTP 500 Internal Server Error**: If the API returns a `500` status code, the client will retry the request. This is demonstrated in `test_retrying_status_errors_doesnt_leak` and `test_retries_taken` in `tests/test_client.py`, where a `httpx.Response(500)` is mocked.
2. **Request Timeout Exception**: If a `httpx.TimeoutException` occurs, the client will also retry
… output trimmedwatcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'# Full old-style diagnostic output: payloads, tool results, telemetry, timestamps.
# This reprints the captured run; it does not call the model or MCP again.
for event in events:
print('event', repr(event))event AgentEvent(type='run_started', message='Agent run started', payload={'prompt': 'Use DeepWiki ask_question on openai/openai-python. Concisely explain retry eligibility, backoff, and streaming cleanup. Name the core classes and methods and give two failure tests. Call exactly one MCP tool unless it fails.'}, timestamp=1786536102.651762)
event AgentEvent(type='mcp_attached', message='MCP server attached: deepwiki', payload={'server': 'deepwiki'}, timestamp=1786536102.655424)
event AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 3, 'iteration': 1}, timestamp=1786536102.656594)
event AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 1297, 'completion_tokens': 73, 'total_tokens': 1370, 'cache_read_input_tokens': 0, 'cache_creation_input_tokens': 0}, 'iteration': 1}, timestamp=1786536104.428396)
event AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 1540, 'completion_tokens': 137, 'total_tokens': 1677, 'cache_read_input_tokens': 0, 'cache_creation_input_tokens': 0}, 'iteration': 1}, timestamp=1786536105.699116)
event AgentEvent(type='agent_decision', message='I am about to use the `deepwiki__ask_question` tool to query the `openai/openai-python` repository for specific implementation details regarding error handling and resource management. I am looking for clarity on retry logic and streaming cleanup, building on the initial task requirement to identify core classes and failure scenarios.', payload={'summary': 'I am about to use the `deepwiki__ask_question` tool to query the `openai/openai-python` repository for specific implementation details regarding error handling and resource management. I am l… (580 more characters)
event AgentEvent(type='tool_group_started', message='Tool group started', payload={'group_id': 'tool_group_1', 'iteration': 1, 'tool_count': 1, 'tools': [{'name': 'deepwiki__ask_question', 'call_id': 'call_1_1'}]}, timestamp=1786536105.700466)
event AgentEvent(type='tool_called', message='Tool called: deepwiki__ask_question', payload={'tool': 'deepwiki__ask_question', 'call_id': 'call_1_1', 'group_id': 'tool_group_1', 'arguments': {'repoName': 'openai/openai-python', 'question': 'Explain retry eligibility, backoff, and streaming cleanup in this library. Name the core classes and methods involved and provide two examples of failure tests (scenarios/error types).'}, 'iteration': 1}, timestamp=1786536105.701344)
event AgentEvent(type='tool_output_started', message='Tool output started: deepwiki__ask_question', payload={'tool': 'deepwiki__ask_question', 'call_id': 'call_1_1', 'group_id': 'tool_group_1', 'attempt': 1, 'buffered': False, 'iteration': 1}, timestamp=1786536105.702649)
event AgentEvent(type='tool_output_delta', message='Tool output: deepwiki__ask_question', payload={'tool': 'deepwiki__ask_question', 'call_id': 'call_1_1', 'group_id': 'tool_group_1', 'chunk': 'This library handles retry eligibility, exponential backoff, and streaming cleanup primarily within the `BaseClient` and `AsyncAPIClient` classes. Streaming cleanup is specifically managed by the `AsyncStream` class. \n\n## Retry Eligibility and Backoff\n\nThe core logic for determining retry eligibility and calculating backoff is found in the `BaseClient` class. \n\n### Core Classes and Methods\n\n*… (3,885 more characters)
event AgentEvent(type='tool_completed', message='Tool completed: deepwiki__ask_question', payload={'tool': 'deepwiki__ask_question', 'call_id': 'call_1_1', 'group_id': 'tool_group_1', 'output': 'This library handles retry eligibility, exponential backoff, and streaming cleanup primarily within the `BaseClient` and `AsyncAPIClient` classes. Streaming cleanup is specifically managed by the `AsyncStream` class. \n\n## Retry Eligibility and Backoff\n\nThe core logic for determining retry eligibility and calculating backoff is found in the `BaseClient` class. \n\n### Core Classes and Methods\n\n… (3,956
… output trimmed# Human-facing progress only. The complete low-level trace remains in `events`.
progress_types = {
'skills_selected', 'agent_decision', 'tool_called', 'agent_observation',
'tool_failed', 'run_failed', 'run_completed',
}
def progress_line(event):
return (
format_event_line(event)
or f'{event.type:<20} {event.display_message}'.rstrip()
)
progress_transcript = [
progress_line(event) for event in events if event.type in progress_types
]
print('\n'.join(progress_transcript))
print('\nRaw event counts:', dict(event_counts))agent_decision I am about to use the `deepwiki__ask_question` tool to query the `openai/openai-python` repository for specific implementation details regarding error handling…
⚙ deepwiki__ask_question(repoName="openai/openai-python", question="Explain retry eligibility, backoff, …) …
agent_observation Asked question openai/openai-python.
run_completed Agent run completed
Raw event counts: {'run_started': 1, 'mcp_attached': 1, 'step_started': 2, 'usage_tick': 3, 'agent_decision': 1, 'tool_group_started': 1, 'tool_called': 1, 'tool_output_started': 1, 'tool_output_delta': 1, 'tool_completed': 1, 'tool_group_completed': 1, 'agent_observation': 1, 'text_delta': 372, 'final_answer': 1, 'run_completed': 1}usage = completed_payload.get('usage', {})
token_report = {
'prompt_tokens': usage.get('prompt_tokens', 0),
'completion_tokens': usage.get('completion_tokens', 0),
'total_tokens': usage.get('total_tokens', 0),
'cache_read_input_tokens': usage.get('cache_read_input_tokens', 0),
'cache_creation_input_tokens': usage.get('cache_creation_input_tokens', 0),
}
run_report = {
'wall_seconds': round(agent_seconds, 2),
'event_counts': dict(event_counts),
'usage': token_report,
'final_output_chars': len(str(completed_payload.get('output', ''))),
'tool_telemetry': tool_telemetry,
}
print(json.dumps(run_report, indent=2, default=str))
print('\nFINAL ANSWER\n')
print(completed_payload.get('output', '')){
"wall_seconds": 21.93,
"event_counts": {
"run_started": 1,
"mcp_attached": 1,
"step_started": 2,
"usage_tick": 3,
"agent_decision": 1,
"tool_group_started": 1,
"tool_called": 1,
"tool_output_started": 1,
"tool_output_delta": 1,
"tool_completed": 1,
"tool_group_completed": 1,
"agent_observation": 1,
"text_delta": 372,
"final_answer": 1,
"run_completed": 1
},
"usage": {
"prompt_tokens": 3790,
"completion_tokens": 492,
"total_tokens": 4282,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0
},
"final_output_chars": 1444,
"tool_telemetry": [
{
"tool": "deepwiki__ask_question",
"output_chars": 3879,
"model_output_chars": 3879,
"model_output_reduced": false,
"metadata": {
"server": "deepwiki",
"ok": true,
"is_error": false,
"output_schema": {
"properties": {
"result": {
… 38 more linesThis local deterministic test proves that a 100,000-character result stays complete while live deltas remain bounded and the model-visible copy is capped.
class OneToolThenAnswer:
def __init__(self):
self.calls = 0
def complete(self, **kwargs):
self.calls += 1
if self.calls == 1:
return LLMResponse(
content='',
tool_calls=[ToolCall(name='large_result', arguments={})],
usage={'prompt_tokens': 10, 'completion_tokens': 2, 'total_tokens': 12},
)
return LLMResponse(
content='Large result processed.',
usage={'prompt_tokens': 20, 'completion_tokens': 4, 'total_tokens': 24},
)
large_text = '0123456789' * 10_000
large_agent = Agent(
llm=OneToolThenAnswer(),
tools=[FunctionTool.from_callable(lambda: large_text, name='large_result')],
auto_use_skills=False,
max_tool_output_chars=4_000,
max_tool_output_group_chars=4_000,
)
large_events = list(large_agent.stream('Run the large result tool.'))
large_deltas = [e.payload['chunk'] for e in large_events if e.type == 'tool_output_delta']
large_completed = next(e for e in large_events if e.type == 'tool_completed')
print({
'canonical_chars': large_completed.payload['output_chars'],
'model_chars': large_completed.payload['model_output_chars'],
'model_output_reduced': large_completed.payload['model_output_reduced'],
'delta_count': len(large_deltas),
'largest_delta': max(map(len, large_deltas)),
'reassembled_complete': ''.join(large_deltas) == large_text,
})watcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'{'canonical_chars': 100000, 'model_chars': 4000, 'model_output_reduced': True, 'delta_count': 7, 'largest_delta': 16384, 'reassembled_complete': True}Provider failures still raise to the caller, but consumers receive a terminal run_failed event first.
class BrokenLLM:
def complete(self, **kwargs):
raise ConnectionError('simulated provider outage')
failure_events = []
stream = Agent(llm=BrokenLLM(), auto_use_skills=False).stream('test failure')
try:
while True:
event = next(stream)
failure_events.append(event)
print(event.type, event.payload)
except StopIteration:
pass
except ConnectionError as exc:
print('Expected exception:', exc)
failed = [event for event in failure_events if event.type == 'run_failed']
assert len(failed) == 1
assert failed[0].payload['retryable'] is Truerun_started {'prompt': 'test failure'}
step_started {'tool_count': 0, 'iteration': 1}
llm_retry {'attempt': 1, 'error': 'simulated provider outage', 'delay': 0.548}
llm_retry {'attempt': 2, 'error': 'simulated provider outage', 'delay': 1.048}
run_failed {'error_type': 'ConnectionError', 'error': 'simulated provider outage', 'retryable': True}
Expected exception: simulated provider outageFuzzy catalog search is explicit. Runtime auto-activation only uses authored trigger phrases, preventing unrelated prompts and tool bundles from inflating every model turn.
skill_agent = Agent(llm=OneToolThenAnswer())
mcp_prompt = 'Use DeepWiki MCP to analyze retry behavior in a repository.'
trigger_prompt = 'Please debug this production bug and plan this feature.'
print('MCP prompt skills:', [s.id for s in skill_agent._selected_skills(mcp_prompt)])
print('Authored-trigger skills:', [s.id for s in skill_agent._selected_skills(trigger_prompt)])
print('Explicit fuzzy search:', [s.id for s in skill_agent.search_skills('database')[:5]])
assert skill_agent._selected_skills(mcp_prompt) == []
assert skill_agent._selected_skills(trigger_prompt)MCP prompt skills: [] Authored-trigger skills: ['code-workflow-assistant'] Explicit fuzzy search: ['database-architect', 'notion-workspace-manager', 'mcp-server-builder', 'backup-manager']
The stress test closest to a real user session. One agent carries 12 local tools, the
live DeepWiki MCP (HTTP), and a second in-process MCP (CRM). Deferred tool loading is
on, so only a small core is advertised up front — the model must tool_search to discover
and load the rest. Reads run in parallel; the whole run streams, and every AgentEvent is
recorded verbatim with a per-type tally, token usage, and estimated cost.
import json
from collections import Counter
from shipit_agent import Agent, MCPServer, MCPTool
from shipit_agent.policies import RetryPolicy
from shipit_agent.tools.base import ToolOutput
from shipit_agent.tools.tool_search import ToolSearchTool
from shipit_agent.tools.ask_user.ask_user_tool import AskUserTool
# ── 12 local tools (read-heavy, so they fan out in parallel) ──────────────
class T:
def __init__(self, name, desc, params, required, fn, read_only=True):
self.name, self.description = name, desc
self._p, self._r, self._fn, self.read_only = params, required, fn, read_only
def schema(self):
return {'type': 'function', 'function': {'name': self.name,
'description': self.description,
'parameters': {'type': 'object', 'properties': self._p, 'required': self._r}}}
def run(self, context, **kw):
return ToolOutput(text=self._fn(**kw), metadata={})
ORDERS = [
{'id': 'A-1', 'customer': 'ACME', 'value_eur': 750, 'status': 'open'},
{'id': 'A-2', 'customer': 'ACME', 'value_eur': 1200, 'status': 'open'},
{'id': 'B-1', 'customer': 'Globex', 'value_eur': 90, 'status': 'open'},
]
local_tools = [
ToolSearchTool(),
AskUserTool(),
T('weather_lookup', 'Current weather for a city.', {'city': {'type': 'string'}}, ['city'],
lambda city, **_: f'Weather in {city}: 21C, clear.'),
T('calculator', 'Evaluate arithmetic.', {'expression': {'type': 'string'}}, ['expression'],
lambda expression, **_: str(eval(expression, {'__builtins__': {}}, {})
if set(expression) <= set('0123456789+-*/(). ') else 'err')),
T('orders_db_query', 'Query open orders by min value (EUR).', {'min_value_eur': {'type': 'number'}}, [],
lambda min_value_eur=0, **_: json.dumps([o for o in ORDERS if o['value_eur'] >= float(min_value_eur or 0)])),
T('currency_convert', 'Convert EUR to USD.', {'amount': {'type': 'number'}}, ['amount'],
lambda amount, **_: f'{float(amount) * 1.08:.2f} USD'),
T('timezone_lookup', 'Local time in a city.', {'city': {'type': 'string'}}, ['city'],
lambda city, **_: f'Local time in {city}: 14:30'),
T('unit_convert', 'Convert km to miles.', {'km': {'type': 'number'}}, ['km'],
lambda km, **_: f'{float(km) * 0.621:.1f} miles'),
T('sentiment', 'Classify sentiment of text.', {'text': {'type': 'string'}}, ['text'],
lambda text, **_: 'positive' if 'good' in text.lower() else 'neutral'),
T('word_count', 'Count words.', {'text': {'type': 'string'}}, ['text'],
lambda text, **_: str(len(text.split()))),
T('uppercase', 'Uppercase a string.', {'text': {'type': 'string'}}, ['text'],
lambda text, **_: text.upper()),
T('reverse', 'Reverse a string.', {'text': {'type': 'string'}}, ['text'],
lambda text, **_: text[::-1]),
]
# ── second, in-process MCP server (CRM) ──────────────────────────────────
CRM = {'ACME': {'tier': 'enterprise', 'owner': '[email protected]'}}
TICKETS = {'ACME': [{'id': 'T-77', 'title': 'SSO fails', 'severity': 'high'}]}
crm = MCPServer(name='crm').register_many([
MCPTool(name='crm_lookup_customer', description='Look up a customer record by company.',
input_schema={'type': 'object', 'properties': {'company': {'type': 'string'}},
'required': ['company']},
handler=lambda context, company='', **_: json.dumps(CRM.get(company, {'error': 'unknown'}))),
MCPTool(name='crm_open_tickets', description="List a customer's open support tickets.",
input_schema={'type': 'object', 'properties': {'company': {'type': 'string'}},
'required': ['company']},
handler=lambda context, company='', **_: json.dumps(TICKETS.get(company, []))),
])
heavy_agent = Agent(
llm=new_llm(),
tools=local_tools,
mcps=[crm, new_deepwiki(namespaced=True)], # two MCP servers at once
deferred_tools=True, # core only; discover the rest
parallel_tool_execution=True,
max_tool_concurrency=4,
progress_summaries=True,
max_iterations=12,
auto_use_skills=False,
auto_project_memory=False,
skill_source=None,
retry_policy=RetryPolicy(request_timeout=180.0),
prompt=('You are a precise operations agent. Use tool_search to find tools you need, '
'then call them. Ground every claim in tool output.'),
)
HEAVY_TASK = (
'Look up ACME in the CRM and their open tickets. Check the weather in Berlin. '
'Total the open orders worth at least 500 EUR and convert the total to USD. '
'Then give me a one-paragraph status summary.'
)
counts = Counter()
events = []
for ev in heavy_agent.stream(HEAVY_TASK):
counts[ev.type] += 1
events.append(ev)
# verbatim line for the load-bearing events
if ev.type == 'tool_called':
print(f" called {ev.payload.get('tool')}({json.dumps(ev.payload.get('arguments', {}))[:70]})")
elif ev.type == 'tool_completed':
print(f" -> {ev.payload.get('tool')}: {str(ev.payload.get('output',''))[:70]}")
elif ev.type == 'interactive_request':
print(f" ASK: {(ev.payload.get('payload') or {}).get('question','')}")
elif ev.type == 'run_summary':
print(f"\n{ev.message}")
print('\n=== event tally ===')
for etype in sorted(counts):
print(f' {counts[etype]:>4} {etype}')
tool_names = [e.payload.get('tool') for e in events if e.type == 'tool_called']
summary = next((e.payload for e in events if e.type == 'run_summary'), {})
print('\ntools used:', tool_names)
print('deferred discovery worked:', 'tool_search' in tool_names)
print('both MCP servers reachable:', any(n and n.startswith('crm') for n in tool_names))
print('estimated cost USD:', summary.get('estimated_cost_usd'))
print('total events:', sum(counts.values()))Exception ignored in: <generator object Tracer.trace at 0x115a8ed40>
Traceback (most recent call last):
File "/opt/homebrew/lib/python3.11/site-packages/shipit_watcher/tracer.py", line 174, in trace
with bind(trace_id=trace_id, parent_id=None, result={},
File "/opt/homebrew/Cellar/[email]/3.11.9_1/Frameworks/Python.framework/Versions/3.11/lib/python3.11/contextlib.py", line 158, in __exit__
self.gen.throw(typ, value, traceback)
File "/opt/homebrew/lib/python3.11/site-packages/shipit_watcher/context.py", line 161, in bind
_context.reset(token)
ValueError: <Token var=<ContextVar name='shipit_watcher_context' default=None at 0x11450f4c0> at 0x115cde4c0> was created in a different Context called tool_search({"query": "CRM customer lookup and open tickets"})
-> tool_search: Best tools for 'CRM customer lookup and open tickets' (ranked by relev
called crm_lookup_customer({"company": "ACME"})
-> crm_lookup_customer: {"tier": "enterprise", "owner": "[email protected]"}
called crm_open_tickets({"company": "ACME"})
-> crm_open_tickets: [{"id": "T-77", "title": "SSO fails", "severity": "high"}]
called weather_lookup({"city": "Berlin"})
-> weather_lookup: Weather in Berlin: 21C, clear.
called orders_db_query({})
-> orders_db_query: [{"id": "A-1", "customer": "ACME", "value_eur": 750, "status": "open"}
called tool_search({"query": "]currency convert["})
-> tool_search: Best tools for ']currency convert[' (ranked by relevance):
1. currency
called calculator({"expression": "750 + 1200"})
-> calculator: 1950
called currency_convert({"amount": 1950})
-> currency_convert: 2106.00 USDwatcher: otlp export failed 500: b'{"message":"Internal Server Error","error":"An unknown error occurred"}'=== event tally ===
9 agent_decision
8 agent_observation
1 final_answer
2 mcp_attached
1 run_completed
1 run_started
10 step_started
435 text_delta
1 tool_arguments_rejected
1 tool_call_healed
8 tool_called
8 tool_completed
9 tool_group_completed
9 tool_group_started
8 tool_output_delta
8 tool_output_started
10 usage_tick
tools used: ['tool_search', 'crm_lookup_customer', 'crm_open_tickets', 'weather_lookup', 'orders_db_query', 'tool_search', 'calculator', 'currency_convert']
deferred discovery worked: True
both MCP servers reachable: True
estimated cost USD: None
total events: 529# Full old-style diagnostic output: payloads, tool results, telemetry, timestamps.
# This reprints the captured run; it does not call the model or MCP again.
for event in events:
print('event', repr(event))event AgentEvent(type='run_started', message='Agent run started', payload={'prompt': 'Look up ACME in the CRM and their open tickets. Check the weather in Berlin. Total the open orders worth at least 500 EUR and convert the total to USD. Then give me a one-paragraph status summary.'}, timestamp=1786536193.165395)
event AgentEvent(type='mcp_attached', message='MCP server attached: crm', payload={'server': 'crm'}, timestamp=1786536193.166385)
event AgentEvent(type='mcp_attached', message='MCP server attached: deepwiki', payload={'server': 'deepwiki'}, timestamp=1786536193.1664069)
event AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 2, 'iteration': 1}, timestamp=1786536193.166792)
event AgentEvent(type='usage_tick', message='Usage updated', payload={'usage': {'prompt_tokens': 1766, 'completion_tokens': 26, 'total_tokens': 1792, 'cache_read_input_tokens': 0, 'cache_creation_input_tokens': 0}, 'iteration': 1}, timestamp=1786536193.868237)
event AgentEvent(type='agent_decision', message='Looking for a tool.', payload={'summary': 'Looking for a tool.', 'next_action': 'call_tools', 'tools': [{'name': 'tool_search', 'arguments': {'query': ''}}], 'iteration': 1, 'generated_by_model': False, 'summary_source': 'tool_call'}, timestamp=1786536193.868365)
event AgentEvent(type='tool_group_started', message='Tool group started', payload={'group_id': 'tool_group_1', 'iteration': 1, 'tool_count': 1, 'tools': [{'name': 'tool_search', 'call_id': 'call_1_1'}]}, timestamp=1786536193.8685539)
event AgentEvent(type='tool_arguments_rejected', message='Tool call missing arguments: tool_search', payload={'tool': 'tool_search', 'arguments': {'query': ''}, 'iteration': 1}, timestamp=1786536193.868635)
event AgentEvent(type='tool_group_completed', message='Tool group completed', payload={'group_id': 'tool_group_1', 'iteration': 1, 'tool_count': 1, 'completed_count': 0, 'summary': ''}, timestamp=1786536193.868646)
event AgentEvent(type='step_started', message='LLM completion started', payload={'tool_count': 2, 'iteration': 2}, timestamp=1786536193.868683)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'I'}, timestamp=1786536194.645694)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' will'}, timestamp=1786536194.645956)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' look'}, timestamp=1786536194.646307)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' up'}, timestamp=1786536194.646723)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' AC'}, timestamp=1786536194.646954)
event AgentEvent(type='text_delta', message='', payload={'chunk': 'ME'}, timestamp=1786536194.647143)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' in'}, timestamp=1786536194.648017)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1786536194.648225)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' CRM'}, timestamp=1786536194.648424)
event AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1786536194.648761)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' check'}, timestamp=1786536194.6489658)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' their'}, timestamp=1786536194.6491678)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' open'}, timestamp=1786536194.649718)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' tickets'}, timestamp=1786536194.649966)
event AgentEvent(type='text_delta', message='', payload={'chunk': ','}, timestamp=1786536194.650624)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' check'}, timestamp=1786536194.806614)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' the'}, timestamp=1786536194.806877)
event AgentEvent(type='text_delta', message='', payload={'chunk': ' weather'}, timestamp=1786536194.807105)
event AgentEvent(ty
… output trimmedchecks = {
'deepwiki_direct_ok': direct.metadata.get('ok') is True,
'agent_completed': bool(completed_payload.get('output')),
'one_mcp_call': event_counts['tool_called'] == 1,
'usage_reported': usage.get('total_tokens', 0) > 0,
'canonical_result_complete': all(t['output_chars'] > 0 for t in tool_telemetry),
'large_result_complete': ''.join(large_deltas) == large_text,
'large_model_copy_reduced': large_completed.payload['model_output_reduced'],
'failure_event_emitted': len(failed) == 1,
'no_unrelated_skill_injection': skill_agent._selected_skills(mcp_prompt) == [],
}
for name, passed in checks.items():
print(f'{"PASS" if passed else "FAIL"}: {name}')
assert all(checks.values()), checks
print('\nAll agent, MCP, streaming, token, and failure checks passed.')