autogen

AutoGen Integration Guide

Use mcp-memory-service as the persistent memory backend for AutoGen 0.4+ multi-agent conversations.

Setup

pip install mcp-memory-service autogen-agentchat httpx
MCP_ALLOW_ANONYMOUS_ACCESS=true memory server --http

Context Injection Before Each Turn

The most effective pattern for AutoGen: retrieve relevant memory and inject it into the agent's system message before each conversation turn.

import httpx
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models import OpenAIChatCompletionClient

MEMORY_URL = "http://localhost:8000"


async def retrieve_context(query: str, tags: list[str] | None = None) -> str:
    """Retrieve relevant memory context for injection into system message."""
    payload = {"query": query, "limit": 5}
    if tags:
        payload["tags"] = tags

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MEMORY_URL}/api/memories/search",
            json=payload,
        )
        memories = response.json().get("memories", [])

    if not memories:
        return ""
    return "Relevant context from memory:\n" + "\n".join(f"- {m['content']}" for m in memories)


async def store_finding(content: str, agent_name: str, tags: list[str] | None = None):
    """Store an important finding from an agent."""
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{MEMORY_URL}/api/memories",
            json={
                "content": content,
                "tags": tags or [],
                "memory_type": "observation",
            },
            headers={"X-Agent-ID": agent_name},
        )


# Build memory-aware agents
async def build_memory_agent(name: str, task_description: str) -> AssistantAgent:
    # Pre-load relevant context before conversation starts
    context = await retrieve_context(task_description)

    system_message = f"You are {name}, a helpful AI assistant."
    if context:
        system_message += f"\n\n{context}"

    return AssistantAgent(
        name=name,
        system_message=system_message,
        model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
    )


async def main():
    task = "Analyze the performance characteristics of our REST API"

    researcher = await build_memory_agent("researcher", task)
    analyst = await build_memory_agent("analyst", task)

    team = RoundRobinGroupChat([researcher, analyst], max_turns=4)
    result = await team.run(task=task)

    # Store key findings for future conversations
    for message in result.messages:
        if len(message.content) > 100:  # Store substantive messages
            await store_finding(
                content=message.content[:500],
                agent_name=message.source,
                tags=["analysis", "api-performance"],
            )

asyncio.run(main())