MCP Manager

from qwen_agent.tools import MCPManager

class MCPManager:
    def initConfig(self, config: Dict) -> List[BaseTool]

config = {
    "mcpServers": {
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/allowed/path"]
        },
        "database": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-sqlite", "db.sqlite"]
        },
        "remote-server": {
            "url": "https://mcp-server.example.com",
            "headers": {"Authorization": "Bearer token"}
        }
    }
}

from qwen_agent.agents import FnCallAgent

# MCP server configuration
mcp_config = {
    "mcpServers": {
        "memory": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-memory"]
        }
    }
}

# Create agent with MCP tools
agent = FnCallAgent(
    function_list=[mcp_config],
    llm={'model': 'qwen-max'}
)

messages = [{
    'role': 'user',
    'content': 'Remember that my favorite color is blue'
}]

for response in agent.run(messages):
    print(response[-1].content)

mcp_config = {
    "mcpServers": {
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]
        },
        "web": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-fetch"]
        },
        "memory": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-memory"]
        }
    }
}

agent = FnCallAgent(
    function_list=[mcp_config],
    llm={'model': 'qwen-max'}
)

mcp_config = {
    "mcpServers": {
        "sse-server": {
            "url": "https://mcp.example.com/sse",
            "headers": {
                "Authorization": "Bearer your-token"
            },
            "sse_read_timeout": 300
        }
    }
}

# Set in .env file
API_KEY=your-api-key
DATABASE_URL=postgresql://...