GDELT数据查询服务

cd gdelt-mcp
uv sync

# Development
uv run python server.py

# With FastMCP CLI
uv run fastmcp run server.py

# Interactive testing
uv run fastmcp dev server.py

project_id|private_key|client_email

from fastmcp import Client
from fastmcp.client.auth import BearerAuth

token = f"{project_id}|{private_key}|{client_email}"
client = Client("https://your-server/mcp", auth=BearerAuth(token=token))

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
import httpx

async def create_gdelt_agent():
    """Create a LangChain agent with GDELT tools organized by category."""
    
    # Bearer token format: project_id|private_key|client_email
    token = f"{project_id}|{private_key}|{client_email}"
    
    # Initialize MultiServerMCPClient with remote server
    client = MultiServerMCPClient({
        "gdelt": {
            "transport": "streamable_http",
            "url": "https://your-server/mcp",
            "auth": httpx.BasicAuth(username="", password=token),
        }
    })
    
    # Get usage guide resource
    usage_guide_blobs = await client.get_resources(
        "gdelt",
        uris="gdelt://guide/usage"
    )
    usage_guide = usage_guide_blobs[0].as_string()
    
    # Get LangChain tools (these include metadata with tags)
    tools = await client.get_tools(server_name="gdelt")
    
    # Group tools by FastMCP tags
    def group_tools_by_tag(tools):
        categories = {
            'schema': [],
            'query': [],
            'cost': [],
            'cameo': [],
            'guide': []
        }
        for tool in tools:
            # Extract tags from LangChain tool metadata
            if hasattr(tool, 'metadata') and tool.metadata:
                tags = tool.metadata.get('tags', [])
                for tag in tags:
                    if tag in categories:
                        categories[tag].append(tool.name)
        return categories
    
    tool_categories = group_tools_by_tag(tools)
    
    # Format categorized tool list for system prompt
    tools_by_category = "\n\n".join([
        f"**{category.upper()} TOOLS**:\n" + 
        "\n".join([f"- {tool}" for tool in tools])
        for category, tools in tool_categories.items()
        if tools
    ])
    
    # Build comprehensive system prompt
    system_prompt = f"""You are a GDELT data analysis assistant with access to the complete GDELT 2.0 dataset via BigQuery.

IMPORTANT USAGE GUIDE:
{usage_guide}

AVAILABLE TOOLS BY CATEGORY:
{tools_by_category}

Follow the recommended workflow from the usage guide above. Always:
1. Check existing materialized subsets first (list_materialized_subsets)
2. Create materialized subsets for iterative analysis (create_materialized_subset)
3. Use appropriate date filters in all queries
4. Start with Events table (smallest) before querying larger tables"""
    
    # Create agent with gpt-5.1
    agent = create_agent(
        "openai:gpt-5.1",
        tools,
        system_prompt=system_prompt
    )
    
    return agent

# Use the agent
agent = await create_gdelt_agent()
result = await agent.ainvoke({
    "messages": "Analyze protests in the United States during the first week of January 2025"
})
print(result)

from mcp import Client
import httpx

# Bearer token format: project_id|private_key|client_email
token = f"{project_id}|{private_key}|{client_email}"
server_url = "https://gdelt-mcp.fastmcp.app/mcp"
auth = httpx.BasicAuth(username="", password=token)

client = Client(server_url, auth=auth)

async with client:
    # Get usage guide resource
    usage_guide = await client.read_resource("gdelt://guide/usage")
    print(f"Usage guide: {usage_guide[:200]}...")
    
    # List all tools
    tools = await client.list_tools()
    print(f"Available tools: {[t.name for t in tools]}")
    
    # Filter tools by tag
    schema_tools = [
        tool for tool in tools 
        if hasattr(tool, 'meta') and tool.meta and
           tool.meta.get('_fastmcp', {}) and
           'schema' in tool.meta.get('_fastmcp', {}).get('tags', [])
    ]
    
    query_tools = [
        tool for tool in tools 
        if hasattr(tool, 'meta') and tool.meta and
           tool.meta.get('_fastmcp', {}) and
           'query' in tool.meta.get('_fastmcp', {}).get('tags', [])
    ]
    
    cost_tools = [
        tool for tool in tools 
        if hasattr(tool, 'meta') and tool.meta and
           tool.meta.get('_fastmcp', {}) and
           'cost' in tool.meta.get('_fastmcp', {}).get('tags', [])
    ]
    
    print(f"Found {len(schema_tools)} schema tools")
    print(f"Found {len(query_tools)} query tools")
    print(f"Found {len(cost_tools)} cost optimization tools")

# Step 1: Create subset once (filters data, small cost)
create_materialized_subset(
    source_table="events",
    subset_name="ukraine_jan2025",
    where_clause="SQLDATE BETWEEN 20250101 AND 20250131 AND (Actor1CountryCode = 'UKR' OR Actor2CountryCode = 'UKR')",
    select_fields="SQLDATE, Actor1Name, Actor2Name, EventCode",
    description="Ukraine events January 2025"
)

# Step 2: Query subset multiple times (near-free)
query_materialized_subset(
    subset_name="ukraine_jan2025",
    where_clause="EventCode LIKE '19%'",  # Military events
    limit=1000
)

query_materialized_subset(
    subset_name="ukraine_jan2025",
    where_clause="GoldsteinScale < -5",  # High conflict
    limit=500
)

query_events(
    where_clause="EventRootCode = '19' AND SQLDATE >= 20250101",
    select_fields="SQLDATE, Actor1Name, Actor2Name, EventCode",
    limit=100
)

query_events(
    where_clause="EventRootCode = '14' AND Actor1CountryCode = 'USA' AND SQLDATE >= 20250101",
    limit=50
)

# First get events
events = query_events(
    where_clause="Actor1Name LIKE '%Biden%' AND SQLDATE >= 20250101",
    limit=10
)

# Then get mentions for specific event
mentions = query_eventmentions(
    where_clause=f"GLOBALEVENTID = {events[0]['GLOBALEVENTID']}"
)

query_gkg(
    where_clause="Themes LIKE '%ENV_CLIMATECHANGE%' AND DATE >= 20250101000000",
    select_fields="DATE, Themes, V2Locations, V2Tone",
    limit=100
)