Pydantic AI
Connect a Pydantic AI agent to Edison Watch over streamable HTTP, and keep a stable session per conversation so data-leak protection holds across every turn.
Pydantic AI connects to Edison with MCPToolset, which resolves the streamable-HTTP transport from your connection URL. The URL carries your API key, so no auth header is needed.
This page targets Pydantic AI v2+, where MCPToolset replaced the removed MCPServerStreamableHTTP class.
pip install "pydantic-ai-slim[mcp,openai]"import asyncio
import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
toolset = MCPToolset(os.environ["EDISON_MCP_URL"])
agent = Agent("openai:gpt-5.6-luna", toolsets=[toolset])
async def main() -> None:
result = await agent.run("List my available tools.")
print(result.output)
asyncio.run(main())Set EDISON_MCP_URL to your connection URL, e.g. https://mcp.edison.watch/mcp/<your-api-key>/?client=pydantic-ai. The toolset opens and closes its connection around the run automatically; wrap it in async with agent: if you want to control the connection lifecycle explicitly.
Keep a stable session across turns to preserve data-leak protection
Send a stable x-edison-conversation-id header on every turn of the same conversation. That header is what keeps Edison's data-leak protection intact across a multi-turn run: Edison tracks lethal-trifecta risk per session, so if each turn looks like a brand-new session, that protection resets - and a later turn can leak data that the accumulated risk should have blocked.
Hosted clients (Claude Code, VS Code) send it automatically. For a custom Pydantic AI agent, set the header yourself, keyed to your own conversation or thread id:
import asyncio
import os
import uuid
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPToolset
def build_agent(conversation_id: str) -> Agent:
toolset = MCPToolset(
os.environ["EDISON_MCP_URL"],
headers={"x-edison-conversation-id": conversation_id},
)
return Agent("openai:gpt-5.2", toolsets=[toolset])
async def main() -> None:
# One stable id for the whole conversation - reuse it on every turn.
conversation_id = f"conv-{uuid.uuid4()}"
agent = build_agent(conversation_id)
async with agent:
first = await agent.run("Summarize my latest support tickets.")
# Same id -> same session: the risk flags from reading tickets still apply here.
second = await agent.run(
"Now email that summary to my manager.",
message_history=first.new_messages(),
)
print(second.output)
asyncio.run(main())Without a stable x-edison-conversation-id, each connection is treated as a fresh session that starts with empty risk state - so risk accumulated on an earlier turn won't be there to block a later exfiltration. The ?client= label is only a dashboard tag, not a session key. Use a unique id per conversation (a UUID is ideal); ids are scoped to your API key, so don't reuse one string for two different conversations.
Optional: the encrypted-secrets header
For servers with zero-knowledge-encrypted secrets, add x-edison-secret-key to headers - alongside x-edison-conversation-id if you're sending both:
toolset = MCPToolset(
os.environ["EDISON_MCP_URL"],
headers={
"x-edison-conversation-id": conversation_id,
"x-edison-secret-key": os.environ["EDISON_SECRET_KEY"],
},
)headers and http_client are mutually exclusive - pass one or the other, not both.
LangChain / LangGraph
Connect a LangChain or LangGraph agent (Python or JS) to Edison Watch using langchain-mcp-adapters, keeping a stable session per conversation so data-leak protection holds across every turn.
LlamaIndex
Connect a LlamaIndex agent to Edison Watch using BasicMCPClient and McpToolSpec, keeping a stable session per conversation so data-leak protection holds across every turn.