agent-switch

agent-switch

Unified abstraction layer for agent SDKs (deepagents, Qcoder SDK, etc.)

agent-switch gives your business code a single, stable API — create_agent + run / stream — so you can switch underlying agent frameworks (deepagents, qcoder, …) without touching your upper-layer types or call sites.

Features

• One API, many backends: create_agent(AgentBackend.DEEPAGENTS | "qcoder", config) returns an adapter exposing the same run(input) -> AgentResponse and stream(input) -> AsyncIterator[AgentChunk].
• Rich, validated type system: AgentConfig, AgentMessage, AgentTool, AgentSkillsConfig, AgentMcpConfig, AgentSubagent, AgentChunk, AgentResponse, … (Pydantic v2, extra="forbid").
• Hooks lifecycle: 12 async hook events (beforeAgent … afterStop) with BLOCK / MODIFY outcomes and a per-AgentConfig hook list.
• Structured logging: redacts secrets, summarizes config/input, Dev & JSON formatters, configured only for the agent_switch namespace (no global handlers).
• Lazy dependencies: deepagents is imported only when the deepagents backend is actually used; import agent_switch never requires it.

Installation

pip install agent-switch
# or with extras
pip install "agent-switch[deepagents]"    # deepagents backend
pip install "agent-switch[qcoder]"        # qcoder backend (qoder-agent-sdk)
pip install "agent-switch[all]"

The qcoder backend runs the real qoder-agent-sdk, which spawns the qodercli CLI. Install the CLI and log in once (qodercli auth) before use.

Quick start

from agent_switch import AgentConfig, AgentMessage, MessageRole, create_agent, AgentBackend

# qcoder runs on the real qoder-agent-sdk (needs `qodercli` installed & logged in)
config = AgentConfig(system_prompt="Be concise.")
agent = create_agent(AgentBackend.QCODER, config)

response = agent.run("Tell me a joke")
print(response.content)

async def demo_stream() -> None:
    async for chunk in agent.stream("Hello"):
        print(chunk.delta_content, end="")

Run the demos

python -m examples                         # unified API entry-point demo (all call styles)
python -m examples.basic_usage             # DEEPAGENTS + QCODER sync run & QCODER stream
python -m examples.deepseek_flash_usage    # DeepSeek Flash via env config (needs DEEPSEEK_API_KEY)

Reusable hook implementations live in examples/hooks.py (audit logging, rate limiting, sensitive-word blocking, context injection) — import them in your own entry point and pass them to AgentConfig(hooks=[...]).

## DeepAgents + `extra["model"]`

For the real `deepagents` backend, pass a pre-built LangChain `ChatModel` through
`AgentConfig.extra["model"]` — it takes priority over `AgentModel`:

```python
from langchain_deepseek import ChatDeepSeek
from agent_switch import AgentConfig, create_agent, AgentBackend

model = ChatDeepSeek(model="deepseek-v4-flash", api_key="sk-...")
config = AgentConfig(
    system_prompt="You are a helpful assistant.",
    tools=[AgentTool(name="search", handler=my_search_tool)],
    extra={"model": model},  # ← pre-built ChatModel wins
)
agent = create_agent(AgentBackend.DEEPAGENTS, config)
response = agent.run("What is the weather in Paris?")

Alternatively let agent-switch build the model: AgentModel(name=..., api_key=..., base_url=...) maps to langchain.chat_models.init_chat_model, while a bare AgentModel(name="openai:gpt-4o-mini") passes the string straight through.

Qcoder ↔ agent-switch mapping

The qcoder backend runs on the real qoder-agent-sdk (which drives the qodercli CLI). It supports message format normalization, the unified hooks lifecycle, streaming, tools / skills / MCP configuration, and session identity.

Message normalization

Input direction (AgentMessage → qoder CLI wire format, via agent_switch.backends.qcoder.mapping.agent_messages_to_qoder_wire):

Output direction (qoder SDK Message → AgentMessage):

Hooks mapping

Session-level events (beforeAgent / beforePrompt / beforeLLM / afterLLM / afterAgent / afterStop) fire at the adapter level once per run / stream, exactly as documented in the Hooks chapter. Call-level events are bridged into the Qoder SDK native hook system:

Only events whose hook class actually overrides the method are registered, so an empty hooks list adds no callbacks to the CLI.

Configuration mapping (AgentConfig → QoderAgentOptions)

Streaming

stream() iterates qoder_agent_sdk.query(prompt=wire_messages, options=...); each SDK AssistantMessage is mapped to one or more AgentChunk (delta_thinking / delta_content / delta_tool_call), and the stream always ends with a chunk carrying is_finish=True. run() wraps the same async flow with asyncio.run and returns the terminal ResultMessage as AgentResponse (falling back to the accumulated assistant text if no result message arrives). Token-level partial messages (StreamEvent) are not enabled by default.

Runtime requirements

• pip install "agent-switch[qcoder]" (pulls qoder-agent-sdk, mcp, anyio)
• Install the qodercli CLI and log in once (qodercli auth)
• Sync run() uses asyncio.run internally: calling it inside a running event loop raises RuntimeError — use stream() in async code.

Hooks

from agent_switch import (
    AgentConfig, AgentHookEvent, BaseAgentHooks, HookOutcome, HookResult, create_agent,
)

class AuditHooks(BaseAgentHooks):
    async def before_llm(self, context) -> None:
        print(f"[audit] beforeLLM model={context.model}")

class RateLimitHooks(BaseAgentHooks):
    async def before_prompt(self, context):
        if len(context.messages) > 10:
            return HookResult(outcome=HookOutcome.BLOCK, reason="rate limit exceeded")

# single instance or a list — both are normalized
config = AgentConfig(hooks=[AuditHooks(), RateLimitHooks()])
# or: AgentConfig(hooks=AuditHooks())
agent = create_agent(AgentBackend.QCODER, config)
response = agent.run("hello")

Run/stream trigger six lifecycle events in order: beforeAgent → beforePrompt → beforeLLM → [SDK] → afterLLM → afterAgent → afterStop. Hooks may return HookResult(outcome=BLOCK, reason=...) (raises HookBlockedError) or HookResult(outcome=MODIFY, data={"messages": [...]}) (replaces the prompt messages).

Hooks fire on two layers:

• Agent level (once per agent execution): beforeAgent, beforePrompt and afterAgent — for the deepagents backend these fire inside the SDK, on the graph's before_agent / after_agent entry/exit nodes.
• Call level (once per LLM / tool call inside the agent loop): beforeLLM, afterLLM, beforeTool, afterTool, afterToolError — bridged through an injected AgentHooksMiddleware (wrap_model_call / wrap_tool_call), so they fire at the real model/tool call points (e.g. several times when the agent loops over tools).
• afterStop (reason complete / error) is fired by the adapter at the run / stream boundary — after_agent only runs on the success path, so the adapter re-fires afterAgent(error) + afterStop(error) when the run fails.

For the qcoder backend, the six session-level events fire at the adapter level (one per run / stream), while beforeTool / afterTool / afterToolError / beforePermission / beforeSubagent / afterSubagent are bridged to the Qoder SDK's native hooks (PreToolUse / PostToolUse / PostToolUseFailure / PermissionRequest / SubagentStart / SubagentStop) and fire inside the CLI. beforePermission / beforeSubagent / afterSubagent are declared but not yet bridged for the deepagents backend.

Hooks ↔ deepagents implementation mapping

Implementation details for the deepagents backend:

• DeepAgentsAdapter._build_agent() appends an AgentHooksMiddleware instance to create_deep_agent(middleware=[...]) whenever AgentConfig.hooks is non-empty; it coexists with user middleware passed via config.extra["middleware"].
• The middleware reads the current session ids through a session_provider closure (bound to the adapter's _session_id / _correlation_id), so all contexts share the same session (session_id / correlation_id) as the adapter-level ones.
• before_agent / after_agent are the graph's entry / exit nodes: each fires exactly once per agent execution (sub-agents are separately compiled graphs and do not trigger them). beforePrompt returning MODIFY rewrites the initial state via {"messages": [...]}; BLOCK raises HookBlockedError inside the SDK, aborting the whole run.
• beforeLLM returning MODIFY rewrites the real request via request.override(messages=...).
• Because these events fire inside the SDK, DeepAgentsAdapter sets call_hooks_via_middleware = True and agent_hooks_via_middleware = True so the adapter layer does not fire them a second time; afterStop (and the error path) remain at the adapter, since after_agent never runs when the graph raises.
• The built graph is cached; the cache key includes a fingerprint of the configured hooks, so changing hooks rebuilds the agent instead of reusing a stale graph.

Streaming

async for chunk in agent.stream("hello"):
    if chunk.delta_content:
        print(chunk.delta_content, end="")
    if chunk.delta_thinking:
        print(f"\n[thinking] {chunk.delta_thinking}")

AgentChunk fields: delta_content, delta_thinking, delta_tool_call, is_finish, meta. The stream always ends with a chunk carrying is_finish=True.

Message model

DeepAgents ↔ agent-switch mapping

thinking / meta are not sent to the backend (input direction); they are only extracted on the way back.

License

MIT