Agent Manager

🤖 AgentManager：管理LLM提供者、代理和MCP服务器的统一生态系统。

代理经理 是一个Python包，它为处理大型语言模型（LLM）提供了一个统一的高级系统。它简化了管理提供者凭据、选择模型和构建AI代理。它还支持通过模型上下文协议（MCP）集成外部工具，以实现更高级的工作流。

______________________________________________________________________

✨ 特性

• 通用提供商支持： 无缝连接到各种云LLM提供商（例如。， 谷歌, 开放人工智能, 奥拉玛, 米斯特拉尔, 格罗克)通过单行代码。
• 代理建设： 用最少的设置快速设置基本代理或高级工具驱动的代理。
• MCP集成： 轻松整合MCP服务器的工具，以实现复杂的外部操作。
• 聊天支持： 支持单轮交互和有状态的连续对话。
• UI（可选）： 交互式用户界面，用于探索AgentManager。

______________________________________________________________________

📄 需求

• Python 3.11+

______________________________________________________________________

🛠️ 安装

• 安装不带UI功能的核心包：

  pip install agentmanager

• 要使用交互式UI，请使用/UI\]附加功能进行安装：

  pip install agentmanager[ui]

• 如果您已经安装了旧版本的AgentManager，请升级以获取最新功能：

  pip install --upgrade agentmanager[ui]

______________________________________________________________________

启动用户界面

你可以直接潜入并探索 AgentManager’s 通过在终端中运行以下命令，可以通过交互式UI实现功能：

agentmanager-ui

这将在浏览器中打开一个美观的界面，您可以在其中：

• 管理LLM提供商
• 配置并运行代理
• 添加和使用多个MCP工具

这是一种亲自动手的方式来观察一切。

______________________________________________________________________

ℹ️ 实用程序方法

这 CloudAgentManager 为发现支持的提供者和模型提供了有用的实用方法。

实用程序使用示例

from agentmanager import CloudAgentManager

cloud_agent_manager = CloudAgentManager()

# Get all supported providers
providers = cloud_agent_manager.get_providers()
print("Supported Providers:", providers)
# Output might be: ['OpenAI', 'Google', 'Ollama', ...]

# Get models for a specific provider
google_models = cloud_agent_manager.get_models("ollama")
print("Ollama Models:", ollama_models)
    
# Get API key page link
page_link_for_api_key = cloud_agent_manager.get_provider_key("mistral")
print("Mistral Key Link:", page_link_for_api_key)

______________________________________________________________________

🚀 快速开始

核心功能封装在CloudAgentManager类中。这是一个单回合聊天的基本示例。

单回合聊天示例

此脚本演示初始化LLM、创建基本代理和获取单个响应。

import asyncio
from agentmanager import CloudAgentManager

# The following constants would typically be read from environment variables or a secure vault
# Placeholder values are used for demonstration.
PROVIDER = "google"
API_KEY = "YOUR_API_KEY_HERE" # !!! REPLACE WITH YOUR ACTUAL API KEY !!!
MODEL_NAME = "gemini-2.5-flash"

async def single_chat():
    # 1️⃣ Initialize the CloudAgentManager
    cloud_agent_manager = CloudAgentManager()

    # 2️⃣ Prepare the LLM (handles API key and model validation)
    llm = cloud_agent_manager.prepare_llm(PROVIDER, API_KEY, MODEL_NAME)
        
    # 3️⃣ Prepare the agent (no MCP tools in this example)
    agent, tools = await cloud_agent_manager.prepare_agent(llm)
    
    # 4️⃣ Send message
    user_message = "what is the capital of India?"    
    response_messages = await cloud_agent_manager.get_agent_response(agent, user_message)
    
    for m in response_messages:
        print(m.content)

if __name__ == "__main__":
    asyncio.run(single_chat())

💬 连续聊天循环

用于使用chat_history维护上下文的交互式多回合对话。

import asyncio
from agentmanager import CloudAgentManager
from typing import List

# The following constants would typically be read from environment variables or a secure vault
# Placeholder values are used for demonstration.
PROVIDER = "google"
API_KEY = "YOUR_API_KEY_HERE" # !!! REPLACE WITH YOUR ACTUAL API KEY !!!
MODEL_NAME = "gemini-2.5-flash"

async def chat_loop():
    # 1️⃣ Initialize the CloudAgentManager
    cloud_agent_manager = CloudAgentManager()

    # 2️⃣ Prepare LLM
    llm = cloud_agent_manager.prepare_llm(PROVIDER, API_KEY, MODEL_NAME)
        
    # 3️⃣ Prepare agent (e.g., without MCP tools)
    agent, tools = await cloud_agent_manager.prepare_agent(llm)

    # 4️⃣ Initialize chat history
    # The agent will use this list to maintain context across turns.
    chat_history: List[Any] = []

    # 5️⃣ Terminal loop
    print("\n--- Start Chat ---")
    print("Type 'exit' or 'quit' to end the session.")
    while True:
        user_input = input("You: ").strip()
        
        if user_input.lower() in {"exit", "quit"}:
            print("👋 Goodbye!")
            break

        if not user_input:
            continue

        try:
            # The cloud_agent_manager updates chat_history in place
            new_messages = await cloud_agent_manager.get_agent_response(agent, user_input, chat_history)
            
            for m in new_messages:
                print(f"Agent: {m.content}")
                
        except Exception as e:
            print(f"❌ Agent failed to respond: {e}")

if __name__ == "__main__":
    asyncio.run(chat_loop())

⚙️ 高级：使用MCP工具的代理

如果您的代理需要通过模型上下文协议（MCP）与外部工具交互，您可以在代理准备期间提供MCP服务器配置列表。

每个MCP条目可以包括：

• name （str）：MCP服务器的唯一名称（必填）
• url （str）：您的MCP服务器的URL（必填）
• headers （可选\[Dict\[str，str\]\]）：标头字典，例如。

  {"Authorization": "Bearer XYZ", "X-Custom": "ABC123", ...}

import asyncio
from agentmanager import CloudAgentManager

# The following constants would typically be read from environment variables or a secure vault
# Placeholder values are used for demonstration.
PROVIDER = "google" # !!! REPLACE WITH PROVIDER YOU WANT TO USE !!!
API_KEY = "YOUR_API_KEY_HERE" # !!! REPLACE WITH PROVIDER'S API KEY !!!
MODEL_NAME = "gemini-2.5-flash" # !!! REPLACE WITH MODEL YOU WANT TO USE !!!

# Define multiple MCP configurations (name, url, optional multiple headers)
mcps=[
    {
        "name": "MyFirstServer",
        "url": "MCP_URL_1" # !!! REPLACE WITH YOUR MCP URL !!!
    },
    {
        "name": "MySecondServer",
        "url": "MCP_URL_2", # !!! REPLACE WITH YOUR MCP URL !!!
        # Optional: Custom headers for authentication/routing
        "header": {
            # !!! REPLACE WITH YOUR ACTUAL HEADER NAMES & VALUES !!!
            "FIRST_HEADER_NAME": "FIRST_HEADER_VALUE",
            "SECOND_HEADER_NAME": "SECOND_HEADER_VALUE",
            # Add more headers as needed ...
        }
    },
    # Add more MCP servers as needed ...
]

async def mcp_agent_example():

    # 1️⃣ Initialize the CloudAgentManager
    cloud_agent_manager = CloudAgentManager()

    # 2️⃣ Prepare LLM
    llm = cloud_agent_manager.prepare_llm(PROVIDER, API_KEY, MODEL_NAME)
    
    # 3️⃣ Prepare agent with MCP configuration
    agent, tools = await cloud_agent_manager.prepare_agent(llm, mcps)

    # 4️⃣ Initialize chat history
    # The agent will use this list to maintain context across turns.
    chat_history: List[Any] = []

    # 5️⃣ Terminal loop
    print("\n--- Start Chat ---")
    print("Type 'exit' or 'quit' to end the session.")
    while True:
        user_input = input("You: ").strip()
        
        if user_input.lower() in {"exit", "quit"}:
            print("👋 Goodbye!")
            break

        if not user_input:
            continue

        try:
            # The cloud_agent_manager updates chat_history in place
            new_messages = await cloud_agent_manager.get_agent_response(agent, user_input, chat_history)
            
            for m in new_messages:
                print(f"Agent: {m.content}")
                
        except Exception as e:
            print(f"❌ Agent failed to respond: {e}")

if __name__ == "__main__":
    asyncio.run(mcp_agent_example())

______________________________________________________________________