1# Managed Agents — Python
2 
3> **Bindings not shown here:** This README covers the most common managed-agents flows for Python. If you need a class, method, namespace, field, or behavior that isn't shown, WebFetch the Python SDK repo **or the relevant docs page** from `shared/live-sources.md` rather than guess. Do not extrapolate from cURL shapes or another language's SDK.
4 
5> **Agents are persistent — create once, reference by ID.** Store the agent ID returned by `agents.create` and pass it to every subsequent `sessions.create`; do not call `agents.create` in the request path. The Anthropic CLI is one convenient way to create agents and environments from version-controlled YAML — its URL is in `shared/live-sources.md`. The examples below show in-code creation for completeness; in production the create call belongs in setup, not in the request path.
6 
7## Installation
8 
9```bash
10pip install anthropic
11```
12 
13## Client Initialization
14 
15```python
16import anthropic
17 
18# Default — resolves credentials from the environment:
19# ANTHROPIC_API_KEY, or ANTHROPIC_AUTH_TOKEN, or an `ant auth login` profile.
20# Prefer this for local dev; don't hardcode a key.
21client = anthropic.Anthropic()
22 
23# Explicit API key (only when you must inject a specific key)
24client = anthropic.Anthropic(api_key="your-api-key")
25```
26 
27---
28 
29## Create an Environment
30 
31```python
32environment = client.beta.environments.create(
33    name="my-dev-env",
34    config={
35        "type": "cloud",
36        "networking": {"type": "unrestricted"},
37    },
38)
39print(environment.id)  # env_...
40```
41 
42---
43 
44## Create an Agent (required first step)
45 
46> ⚠️ **There is no inline agent config.** `model`/`system`/`tools` live on the agent object, not the session. Always start with `agents.create()` — the session only takes `agent={"type": "agent", "id": agent.id}`.
47 
48### Minimal
49 
50```python
51# 1. Create the agent (reusable, versioned)
52agent = client.beta.agents.create(
53    name="Coding Assistant",
54    model="claude-opus-4-8",
55    tools=[{"type": "agent_toolset_20260401", "default_config": {"enabled": True}}],
56)
57 
58# 2. Start a session
59session = client.beta.sessions.create(
60    agent={"type": "agent", "id": agent.id, "version": agent.version},
61    environment_id=environment.id,
62)
63print(session.id, session.status)
64```
65 
66### With system prompt and custom tools
67 
68```python
69import os
70 
71agent = client.beta.agents.create(
72    name="Code Reviewer",
73    model="claude-opus-4-8",
74    system="You are a senior code reviewer.",
75    tools=[
76        {"type": "agent_toolset_20260401"},
77        {
78            "type": "custom",
79            "name": "run_tests",
80            "description": "Run the test suite",
81            "input_schema": {
82                "type": "object",
83                "properties": {
84                    "test_path": {"type": "string", "description": "Path to test file"}
85                },
86                "required": ["test_path"],
87            },
88        },
89    ],
90)
91 
92session = client.beta.sessions.create(
93    agent={"type": "agent", "id": agent.id, "version": agent.version},
94    environment_id=environment.id,
95    title="Code review session",
96    resources=[
97        {
98            "type": "github_repository",
99            "url": "https://github.com/owner/repo",
100            "mount_path": "/workspace/repo",
101            "authorization_token": os.environ["GITHUB_TOKEN"],
102            "branch": "main",
103        }
104    ],
105)
106```
107 
108---
109 
110## Send a User Message
111 
112```python
113client.beta.sessions.events.send(
114    session_id=session.id,
115    events=[
116        {
117            "type": "user.message",
118            "content": [{"type": "text", "text": "Review the auth module"}],
119        }
120    ],
121)
122```
123 
124> 💡 **Stream-first:** Open the stream *before* (or concurrently with) sending the message. The stream only delivers events that occur after it opens — stream-after-send means early events arrive buffered in one batch. See [Steering Patterns](../../shared/managed-agents-events.md#steering-patterns).
125 
126---
127 
128## Stream Events (SSE)
129 
130```python
131import json
132 
133# Stream-first: open stream, then send while stream is live
134with client.beta.sessions.events.stream(
135    session_id=session.id,
136) as stream:
137    client.beta.sessions.events.send(
138        session_id=session.id,
139        events=[{"type": "user.message", "content": [{"type": "text", "text": "..."}]}],
140    )
141    for event in stream:
142        ...  # process events
143 
144# Standalone stream iteration:
145with client.beta.sessions.events.stream(
146    session_id=session.id,
147) as stream:
148    for event in stream:
149        if event.type == "agent.message":
150            for block in event.content:
151                if block.type == "text":
152                    print(block.text, end="", flush=True)
153        elif event.type == "agent.custom_tool_use":
154            # Custom tool invocation — session is now idle
155            print(f"\nCustom tool call: {event.name}")
156            print(f"Input: {json.dumps(event.input)}")
157            # Send result back (see below)
158        elif event.type == "session.status_idle":
159            print("\n--- Agent idle ---")
160        elif event.type == "session.status_terminated":
161            print("\n--- Session terminated ---")
162            break
163```
164 
165---
166 
167## Provide Custom Tool Result
168 
169```python
170client.beta.sessions.events.send(
171    session_id=session.id,
172    events=[
173        {
174            "type": "user.custom_tool_result",
175            "custom_tool_use_id": "sevt_abc123",
176            "content": [{"type": "text", "text": "All 42 tests passed."}],
177        }
178    ],
179)
180```
181 
182---
183 
184## Poll Events
185 
186```python
187events = client.beta.sessions.events.list(
188    session_id=session.id,
189)
190for event in events.data:
191    print(f"{event.type}: {event.id}")
192```
193 
194> ⚠️ **Prefer the SDK over raw `requests`/`httpx`.** If you hand-roll a poll loop, don't assume `timeout=(5, 60)` or `httpx.Timeout(120)` caps total call duration — both are **per-chunk** read timeouts (reset on every byte), so a trickling response can block forever. For a hard wall-clock deadline, track `time.monotonic()` at the loop level and bail explicitly, or wrap with `asyncio.wait_for()`. See [Receiving Events](../../shared/managed-agents-events.md#receiving-events).
195 
196---
197 
198## Full Streaming Loop with Custom Tools
199 
200```python
201import json
202 
203 
204def run_custom_tool(tool_name: str, tool_input: dict) -> str:
205    """Execute a custom tool and return the result."""
206    if tool_name == "run_tests":
207        # Your tool implementation here
208        return "All tests passed."
209    return f"Unknown tool: {tool_name}"
210 
211 
212def run_session(client, session_id: str):
213    """Stream events and handle custom tool calls."""
214    while True:
215        with client.beta.sessions.events.stream(
216            session_id=session_id,
217        ) as stream:
218            tool_calls = []
219            for event in stream:
220                if event.type == "agent.message":
221                    for block in event.content:
222                        if block.type == "text":
223                            print(block.text, end="", flush=True)
224                elif event.type == "agent.custom_tool_use":
225                    tool_calls.append(event)
226                elif event.type == "session.status_idle":
227                    break
228                elif event.type == "session.status_terminated":
229                    return
230 
231        if not tool_calls:
232            break
233 
234        # Process custom tool calls
235        results = []
236        for call in tool_calls:
237            result = run_custom_tool(call.name, call.input)
238            results.append({
239                "type": "user.custom_tool_result",
240                "custom_tool_use_id": call.id,
241                "content": [{"type": "text", "text": result}],
242            })
243 
244        client.beta.sessions.events.send(
245            session_id=session_id,
246            events=results,
247        )
248```
249 
250---
251 
252## Upload a File
253 
254```python
255with open("data.csv", "rb") as f:
256    file = client.beta.files.upload(
257        file=f,
258    )
259 
260# Use in a session
261session = client.beta.sessions.create(
262    agent={"type": "agent", "id": agent.id, "version": agent.version},
263    environment_id=environment.id,
264    resources=[{"type": "file", "file_id": file.id, "mount_path": "/workspace/data.csv"}],
265)
266```
267 
268---
269 
270## List and Download Session Files
271 
272List files the agent wrote to `/mnt/session/outputs/` during a session, then download them.
273 
274```python
275# List files associated with a session
276files = client.beta.files.list(
277    scope_id=session.id,
278    betas=["managed-agents-2026-04-01"],
279)
280for f in files.data:
281    print(f.filename, f.size_bytes)
282    # Download each file and save to disk
283    file_content = client.beta.files.download(f.id)
284    file_content.write_to_file(f.filename)
285```
286 
287> 💡 There's a brief indexing lag (~1–3s) between `session.status_idle` and output files appearing in `files.list`. Retry once or twice if the list is empty.
288 
289---
290 
291## Session Management
292 
293```python
294# Get session details
295session = client.beta.sessions.retrieve(session_id="sesn_011CZxAbc123Def456")
296print(session.status, session.usage)
297 
298# List sessions
299sessions = client.beta.sessions.list()
300 
301# Delete a session
302client.beta.sessions.delete(session_id="sesn_011CZxAbc123Def456")
303 
304# Archive a session
305client.beta.sessions.archive(session_id="sesn_011CZxAbc123Def456")
306```
307 
308---
309 
310## MCP Server Integration
311 
312```python
313# Agent declares MCP server (no auth here — auth goes in a vault)
314agent = client.beta.agents.create(
315    name="MCP Agent",
316    model="claude-opus-4-8",
317    mcp_servers=[
318        {"type": "url", "name": "my-tools", "url": "https://my-mcp-server.example.com/sse"},
319    ],
320    tools=[
321        {"type": "agent_toolset_20260401", "default_config": {"enabled": True}},
322        {"type": "mcp_toolset", "mcp_server_name": "my-tools"},
323    ],
324)
325 
326# Session attaches vault(s) containing credentials for those MCP server URLs
327session = client.beta.sessions.create(
328    agent=agent.id,
329    environment_id=environment.id,
330    vault_ids=[vault.id],
331)
332```
333 
334See `shared/managed-agents-tools.md` §Vaults for creating vaults and adding credentials.
335