generic agent

通用代理

一个全面的Python包，用于Azure OpenAI与MCP（模型上下文协议）服务器和智能代理路由的集成。

特性

🔥 Azure OpenAI集成

• 托管身份验证（Azure托管环境首选）
• 服务主体和交互式身份验证回退
• 密钥保险库集成，用于安全的凭证存储
• 指数回退自动重试
• 全面的日志记录和监控

🏗️ MCP服务器构建器

• 基于FastMCP库构建
• 企业级功能（健康检查、指标、监控）
• 轻松注册工具和资源
• 错误处理和恢复
• 连接池和性能优化

🧠 智能代理

• 使用Azure OpenAI进行智能MCP服务器选择
• 自然语言任务分析
• 缓存和性能优化
• 全面的指标和监控
• 错误处理的回退策略

安装

pip install generic-agent

快速开始

1.设置环境变量

创建一个 .env 包含Azure OpenAI配置的文件：

AZURE_OPENAI_ENDPOINT=https://your-openai-resource.openai.azure.com/
AZURE_OPENAI_API_VERSION=2024-02-15-preview
AZURE_OPENAI_DEFAULT_MODEL=gpt-4
AZURE_USE_MANAGED_IDENTITY=true

2.基本用法

import asyncio
from generic_agent import (
    AzureOpenAIClient,
    MCPServerConstructor,
    MCPServerConfig,
    SmartAgent,
    AgentConfig
)
from generic_agent.config import AzureOpenAIConfig

async def example_tool(data: str, operation: str) -> str:
    """Example tool implementation."""
    return f"Processed {data} with {operation}"

async def main():
    # Create Azure OpenAI configuration
    azure_config = AzureOpenAIConfig.from_environment()

    # Create agent configuration
    agent_config = AgentConfig(
        name="MyAgent",
        openai_config=azure_config
    )

    # Initialize smart agent
    agent = SmartAgent(agent_config)

    # Create MCP server
    server_config = MCPServerConfig(
        name="data-server",
        description="Data processing server",
        capabilities=["data_processing", "transformation"]
    )

    server = MCPServerConstructor(server_config)
    server.add_tool(
        name="process_data",
        description="Process data with various operations",
        handler=example_tool,
        parameters={
            "data": {"type": "string"},
            "operation": {"type": "string"}
        },
        required=["data", "operation"]
    )

    # Register server with agent
    agent.register_server(server)

    # Use intelligent task routing
    task = "I need to clean and validate user input data"
    selection = await agent.select_server(task)

    print(f"Selected server: {selection.selected_server}")
    print(f"Confidence: {selection.confidence}")
    print(f"Reasoning: {selection.reasoning}")

    await agent.close()

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

建筑

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   Smart Agent   │    │  Azure OpenAI    │    │   MCP Servers   │
│                 │    │     Client       │    │                 │
│ • Task Analysis │◄──►│ • Managed ID     │    │ • File Server   │
│ • Server Select │    │ • Key Vault      │    │ • Data Server   │
│ • Caching      │    │ • Retry Logic    │    │ • API Server    │
│ • Monitoring   │    │ • Monitoring     │    │ • Custom...     │
└─────────────────┘    └──────────────────┘    └─────────────────┘

认证

该软件包支持遵循Azure最佳实践的多种身份验证方法：

管理身份（推荐）

对于Azure托管的应用程序（应用程序服务、容器应用程序、功能、VM）：

azure_config = AzureOpenAIConfig(
    endpoint="https://your-openai.openai.azure.com/",
    use_managed_identity=True
)

服务主体

对于CI/CD管道和自动化服务：

AZURE_CLIENT_ID=your-client-id
AZURE_CLIENT_SECRET=your-client-secret
AZURE_TENANT_ID=your-tenant-id

密钥库集成

将API密钥安全存储在Azure密钥保管库中：

AZURE_KEY_VAULT_URL=https://your-keyvault.vault.azure.net/
AZURE_OPENAI_API_KEY_SECRET_NAME=openai-api-key

MCP服务器创建

基本服务器

from generic_agent import MCPServerConstructor, MCPServerConfig

# Configure server
config = MCPServerConfig(
    name="my-server",
    description="My custom MCP server",
    host="localhost",
    port=8000,
    capabilities=["data_processing", "file_operations"]
)

# Create server
server = MCPServerConstructor(config)

# Add tools
async def my_tool(param1: str, param2: int) -> str:
    return f"Processed {param1} with value {param2}"

server.add_tool(
    name="my_tool",
    description="Process data with parameters",
    handler=my_tool,
    parameters={
        "param1": {"type": "string", "description": "Input string"},
        "param2": {"type": "integer", "description": "Input number"}
    },
    required=["param1", "param2"]
)

# Add health checks
async def health_check() -> bool:
    # Your health check logic
    return True

server.add_health_check(health_check)

# Build and start
await server.build()
await server.start()

企业服务器功能

# Add monitoring
server.add_health_check(custom_health_check)

# Add resources
server.add_resource(
    uri="data://processed",
    name="Processed Data",
    description="Access to processed data store",
    mime_type="application/json"
)

# Get metrics
metrics = server.metrics
server_info = server.get_server_info()

智能代理功能

任务分析

agent = SmartAgent(config)

# Analyze any task
analysis = await agent.analyze_task("Backup the user database")
print(f"Task type: {analysis.task_type}")
print(f"Complexity: {analysis.complexity_score}/10")
print(f"Required capabilities: {analysis.required_capabilities}")

服务器选择

# Intelligent server selection
selection = await agent.select_server("Process customer data and generate report")
print(f"Best server: {selection.selected_server}")
print(f"Confidence: {selection.confidence}")
print(f"Reasoning: {selection.reasoning}")

# Alternative recommendations
recommendations = agent.get_server_recommendations("Encrypt sensitive files")
for server_name, confidence in recommendations:
    print(f"{server_name}: {confidence:.2f}")

性能监控

# Get performance metrics
metrics = agent.get_performance_metrics()
print(f"Total selections: {metrics['total_selections']}")
print(f"Cache hit rate: {metrics['cache_hit_rate']:.2%}")
print(f"Average selection time: {metrics['avg_selection_time']:.3f}s")

# Clear cache if needed
agent.clear_cache()

例子

基本用法

看 examples/basic_usage.py 作为一个完整的基本示例。

企业模式

看 examples/enterprise_example.py 高级企业模式包括：

• 性能监控和警报
• 错误处理和恢复
• 安全和合规功能
• 生产就绪配置

配置

Azure OpenAI配置

from generic_agent.config import AzureOpenAIConfig

config = AzureOpenAIConfig(
    endpoint="https://your-openai.openai.azure.com/",
    api_version="2024-02-15-preview",
    default_model="gpt-4",
    max_retries=3,
    timeout=30,
    use_managed_identity=True,
    key_vault_url="https://your-keyvault.vault.azure.net/",
    api_key_secret_name="openai-api-key"
)

代理配置

from generic_agent.config import AgentConfig

config = AgentConfig(
    name="MyAgent",
    description="Intelligent task routing agent",
    openai_config=azure_config,
    server_selection_model="gpt-4",
    selection_temperature=0.1,
    max_server_selection_retries=3,
    cache_selection_results=True,
    cache_ttl_seconds=300
)

MCP服务器配置

from generic_agent.config import MCPServerConfig

config = MCPServerConfig(
    name="my-server",
    description="Custom server description",
    host="localhost",
    port=8000,
    capabilities=["capability1", "capability2"],
    tools={"tool1": "description"},
    resources={"resource1": "description"},
    max_connections=100,
    timeout=30
)

错误处理

该软件包提供全面的错误处理：

from generic_agent.exceptions import (
    GenericAgentError,
    AzureOpenAIError,
    MCPServerError,
    AgentError,
    AuthenticationError
)

try:
    selection = await agent.select_server(task)
except AgentError as e:
    print(f"Agent error: {e}")
    print(f"Task: {e.task}")
    print(f"Selected server: {e.selected_server}")
except AzureOpenAIError as e:
    print(f"Azure OpenAI error: {e}")
    print(f"Status code: {e.status_code}")
    print(f"Request ID: {e.request_id}")

日志记录

设置结构化日志记录：

from generic_agent.utils import setup_logging

setup_logging(
    level="INFO",
    format_json=True,
    log_file="agent.log"
)

最佳实践

安全

• 在Azure中运行时使用托管身份
• 将机密存储在Azure密钥库中
• 启用全面的审计日志记录
• 实施适当的错误处理

演出

• 为服务器选择启用缓存
• 监控响应时间和错误率
• 对数据库使用连接池
• 为外部服务实施断路器

监控

• 为所有服务器设置健康检查
• 监控性能指标
• 对性能下降实施警报
• 跟踪任务成功/失败率

发展

• 使用特定于环境的配置
• 实施全面测试
• 始终如一地遵循async/await模式
• 全程使用类型提示

贡献

1. 分叉存储库
2. 创建要素分支
3. 进行更改
4. 添加测试
5. 提交拉取请求

许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

支持

对于问题和疑问：

• 检查示例目录以了解使用模式
• 查看API文档
• 在GitHub上打开一个问题

更新日志

v0.1.0

• 初始版本
• 具有托管身份支持的Azure OpenAI客户端
• 具有FastMCP集成的MCP服务器构造函数
• 具有智能服务器选择功能的智能代理
• 全面的错误处理和监控
• 企业级示例和文档这是一个能够与任何MCP服务器交互的通用代理。