网页抓取编排系统

User Input
   ↓
Agno Orchestrator (preprocesses & coordinates)
   ↓
MCP Tools Framework (manages execution)
   ↓
Primary: Tavily API (fast, structured extraction)
   ↓ (on failure)
Fallback: Jina API (semantic understanding)
   ↓
Formatted Response + Execution Logs

git clone https://github.com/yourusername/mcp-server.git
cd mcp-server

pip install -r requirements.txt

# Copy example environment file
cp .env.example .env

# Edit .env with your API keys
# Required:
TAVILY_API_KEY=your_tavily_api_key_here

# Optional:
JINA_API_KEY=your_jina_api_key_here

# Configuration (optional):
LOG_LEVEL=INFO
LOG_FILE=agno_system.log
MAX_RETRIES=3
TIMEOUT_SECONDS=30

python -m src.server

# Basic usage
python -m src.cli "Tell me about Microsoft 2024 report"

# With custom formatting
python -m src.cli "What happened in AI in 2024?" --format markdown

# With more results
python -m src.cli "Latest climate change news" --max-results 10

# Show statistics
python -m src.cli "Search query" --stats

import asyncio
from src.agno_orchestrator import AgnoOrchestrator

async def main():
    # Initialize orchestrator
    orchestrator = AgnoOrchestrator()
    
    # Process a request
    response = await orchestrator.process_request(
        user_input="Tell me about Microsoft 2024 report",
        max_results=5
    )
    
    # Format and display
    if response.success:
        formatted = orchestrator.format_response(response, format_type="markdown")
        print(formatted)
    else:
        print(f"Error: {response.error}")

asyncio.run(main())

2024-10-14 15:30:45 UTC - AgnoSystem - INFO - [Tavily_Request] Status: success | Duration: 1250.45ms | Quality: {"confidence": 0.85}

# Fallback triggers when:
1. Primary tool fails (error, timeout)
2. Result confidence below threshold
3. Empty or invalid data returned

# Fallback is logged:
{
  "timestamp": "2024-10-14T15:30:45.123Z",
  "operation": "Fallback_Triggered",
  "from": "Tavily",
  "to": "Jina",
  "reason": "Tavily request timeout"
}

{
  "timestamp": "2024-10-14T15:30:45.123456Z",
  "operation": "Tavily_Request_End",
  "status": "success",
  "duration_ms": 1250.45,
  "error": null,
  "data_quality": {
    "confidence": 0.85
  },
  "metadata": {
    "result_count": 5,
    "search_depth": "basic"
  }
}

mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py              # MCP server entry point
│   ├── cli.py                 # Command-line interface
│   ├── config.py              # Configuration management
│   ├── logging_system.py      # Comprehensive logging
│   ├── agno_orchestrator.py   # Main orchestration logic
│   ├── mcp_tools_integration.py  # Tool framework
│   └── tools/
│       ├── __init__.py
│       ├── tavily_tool.py     # Tavily integration
│       └── jina_tool.py       # Jina integration
├── requirements.txt           # Python dependencies
├── .env.example              # Example environment file
├── .gitignore                # Git ignore rules
└── README.md                 # This file

from src.mcp_tools_integration import BaseTool, ToolResult, ToolStatus

class MyCustomTool(BaseTool):
    def __init__(self, api_key=None):
        super().__init__(name="MyTool", priority=2)  # Lower priority = fallback
        self.api_key = api_key
    
    def is_available(self) -> bool:
        return bool(self.api_key)
    
    async def execute(self, query: str, **kwargs) -> ToolResult:
        # Implement your scraping logic
        try:
            # ... scraping code ...
            return ToolResult(
                status=ToolStatus.SUCCESS,
                data={"results": []},
                confidence=0.8
            )
        except Exception as e:
            return ToolResult(
                status=ToolStatus.FAILURE,
                error=str(e)
            )

from src.tools.my_custom_tool import MyCustomTool

# In AgnoOrchestrator._register_tools()
custom_tool = MyCustomTool(api_key=custom_api_key)
self.framework.register_tool(custom_tool)

# Test CLI
python -m src.cli "test query" --format json

# Test specific queries
python -m src.cli "Microsoft 2024 report" --stats
python -m src.cli "Latest AI news" --max-results 10 --format markdown

stats = orchestrator.get_statistics()
# Returns:
{
  "total_requests": 100,
  "successful_requests": 95,
  "success_rate": 95.0,
  "fallback_count": 12,
  "fallback_rate": 12.0,
  "available_tools": 2,
  "registered_tools": 2
}