AutoGen’s agent framework shines when you can swap models without rewriting agent logic. This tutorial shows how to wire Claude 3.5 Sonnet into AutoGen through n4n.ai’s OpenAI-compatible endpoint, giving you access to Anthropic’s flagship model alongside 240+ others behind a single base URL. You’ll build a working multi-agent system that actually runs.
Prerequisites
- Python 3.10+
- An n4n.ai API key (get one at n4n.ai)
- Basic familiarity with AutoGen concepts (agents, conversations, function calling)
Install the dependencies:
pip install pyautogen openai python-dotenvCreate a .env file in your project root:
N4N_API_KEY=your_key_here
N4N_BASE_URL=https://api.n4n.ai/v1Configure the model client
AutoGen 0.2+ uses the OpenAIWrapper (or ChatCompletionClient in newer versions) to talk to any OpenAI-compatible endpoint. Point it at n4n.ai and specify the model identifier.
# config.py
import os
from dotenv import load_dotenv
from autogen import OpenAIWrapper
load_dotenv()
def get_claude_client():
return OpenAIWrapper(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="anthropic/claude-3.5-sonnet",
temperature=0.3,
)The model string anthropic/claude-3.5-sonnet follows n4n.ai’s provider/model convention. You can verify available models by hitting GET /models on the base URL.
Build a minimal two-agent system
Create a researcher agent that gathers information and a writer agent that synthesizes it. Both use the same Claude 3.5 Sonnet client.
# agents.py
from autogen import AssistantAgent, UserProxyAgent
from config import get_claude_client
claude_client = get_claude_client()
researcher = AssistantAgent(
name="researcher",
llm_config={"client": claude_client},
system_message=(
"You are a technical researcher. Given a topic, produce a concise "
"bullet-point summary of key facts, APIs, and best practices. "
"Cite sources when possible. Keep responses under 300 words."
),
)
writer = AssistantAgent(
name="writer",
llm_config={"client": claude_client},
system_message=(
"You are a technical writer. Transform research notes into a clear, "
"well-structured tutorial section with code examples. Use markdown. "
"Target audience: senior engineers. No fluff."
),
)
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="NEVER",
max_consecutive_auto_reply=3,
code_execution_config=False,
)Orchestrate the conversation
AutoGen’s GroupChat manages turn-taking. We’ll seed it with a topic and let the researcher and writer collaborate.
# main.py
from autogen import GroupChat, GroupChatManager
from agents import researcher, writer, user_proxy
def run_pipeline(topic: str):
groupchat = GroupChat(
agents=[user_proxy, researcher, writer],
messages=[],
max_round=6,
)
manager = GroupChatManager(groupchat=groupchat, llm_config={"client": researcher.llm_config["client"]})
user_proxy.initiate_chat(
manager,
message=f"Research and write a tutorial section on: {topic}",
)
return groupchat.messages
if __name__ == "__main__":
messages = run_pipeline("async context managers in Python 3.11+")
for msg in messages:
print(f"[{msg['name']}] {msg['content'][:200]}...")Run it:
python main.pyExpected output (truncated):
[user_proxy] Research and write a tutorial section on: async context managers in Python 3.11+...
[researcher] • Python 3.11 adds `contextlib.AsyncContextManager` base class
• `async with` now supports exception groups via `except*`
• Performance: ~10% faster context manager entry/exit
• New `aclosing()` utility for async generators...
[writer] ## Async context managers in Python 3.11+
Python 3.11 introduces several improvements to async context management...
### Basic pattern
```python
import contextlib
class DatabaseConnection:
async def __aenter__(self):
self.conn = await connect()
return self.conn
async def __aexit__(self, exc_type, exc, tb):
await self.conn.close()…
## Add function calling for live data
Claude 3.5 Sonnet supports function calling through n4n.ai. Equip the researcher with a search tool that hits a real API.
```python
# tools.py
import json
import httpx
from typing import Annotated
from autogen import register_function
async def web_search(query: Annotated[str, "Search query"]) -> str:
"""Search the web and return top 3 result snippets."""
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.get(
"https://api.duckduckgo.com/",
params={"q": query, "format": "json", "no_html": 1, "skip_disambig": 1},
)
data = resp.json()
snippets = [
data.get("Abstract", ""),
data.get("AbstractText", ""),
data.get("RelatedTopics", [{}])[0].get("Text", ""),
]
return json.dumps([s for s in snippets if s][:3])
# Register with the researcher agent
register_function(
web_search,
caller=researcher,
executor=user_proxy,
name="web_search",
description="Search the web for current information",
)Update the researcher’s system message to instruct tool use:
# agents.py (updated researcher)
researcher = AssistantAgent(
name="researcher",
llm_config={"client": claude_client},
system_message=(
"You are a technical researcher. Use the web_search tool to gather "
"current information. Produce a concise bullet-point summary of key "
"facts, APIs, and best practices. Cite sources. Keep responses under 300 words."
),
)Now the researcher will call web_search automatically when the topic demands current data.
Handle streaming and token usage
n4n.ai returns standard OpenAI streaming chunks and includes usage metadata. Wrap the client to capture token counts per agent turn.
# streaming.py
from autogen import OpenAIWrapper
import os
class MeteredClient(OpenAIWrapper):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.total_prompt_tokens = 0
self.total_completion_tokens = 0
def create(self, *args, **kwargs):
response = super().create(*args, **kwargs)
if hasattr(response, "usage") and response.usage:
self.total_prompt_tokens += response.usage.prompt_tokens
self.total_completion_tokens += response.usage.completion_tokens
return response
def create_stream(self, *args, **kwargs):
for chunk in super().create_stream(*args, **kwargs):
if hasattr(chunk, "usage") and chunk.usage:
self.total_prompt_tokens += chunk.usage.prompt_tokens
self.total_completion_tokens += chunk.usage.completion_tokens
yield chunk
# Usage
metered = MeteredClient(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="anthropic/claude-3.5-sonnet",
)Plug metered into your agents’ llm_config instead of the raw client. After a run, metered.total_prompt_tokens and metered.total_completion_tokens give you the full conversation cost.
Fallback behavior
n4n.ai automatically routes around degraded providers. If Anthropic’s API is rate-limited or down, the gateway fails over to another provider serving Claude 3.5 Sonnet (or a compatible model) without changing your code. You’ll see the x-n4n-provider header in responses indicating which upstream served the request.
To observe this, add a response hook:
# debug_fallback.py
import httpx
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
)
async def test_fallback():
resp = await client.chat.completions.create(
model="anthropic/claude-3.5-sonnet",
messages=[{"role": "user", "content": "Say hello"}],
max_tokens=10,
)
print(resp.model_dump_json(indent=2))
# Check resp.headers.get("x-n4n-provider") in raw responseCommon pitfalls
Model identifier mismatch — Use the exact string anthropic/claude-3.5-sonnet. The anthropic/ prefix is required by n4n.ai’s routing layer.
Temperature too high — Claude 3.5 Sonnet performs best for coding tasks at 0.2–0.4. Higher values introduce hallucinated APIs.
Max tokens too low — AutoGen’s default max_tokens may truncate function calling responses. Set max_tokens=4096 in llm_config for complex tool chains.
Streaming with GroupChat — AutoGen’s GroupChatManager doesn’t natively stream. If you need token-by-token output, implement a custom manager that yields from client.create_stream() and handles turn-taking manually.
Next steps
- Add a critic agent that reviews the writer’s output for factual accuracy
- Persist conversation history to a vector store for long-running research tasks
- Use n4n.ai’s routing directives (
x-n4n-prefer-provider,x-n4n-max-latency-ms) to enforce latency SLAs per agent - Swap
anthropic/claude-3.5-sonnetforopenai/gpt-4oormeta-llama/llama-3.1-405bby changing one line — no agent logic changes required
The complete runnable example is available in the n4n.ai examples repository. Clone it, add your API key, and you have a production-ready multi-agent pipeline in under 100 lines of code.