数据中心天气代理

┌─────────────────────────────────────────────────────────────────┐
│                    Data Center Weather Agent                    │
│                                                                 │
│  ┌─────────────┐         ┌───────────────────────────────────┐  │
│  │   Agent     │  HTTP   │         MCP Server                │  │
│  │ (LangGraph) │◄───────►│  ┌────────┐ ┌─────────┐ ┌───────┐ │  │
│  │             │   SSE   │  │ ipify  │ │ip_to_geo│ │weather│ │  |
│  │  5 Nodes    │         │  └───┬────┘ └────┬────┘ └──┬────┘ │  │
│  │  3 Edges    │         │      │           │         │      │  │
│  │  9 State    │         │      ▼           ▼         ▼      │  │
│  │  Fields     │         │    ipify.org  ip-api.com  meteo   │  │
│  └─────────────┘         └───────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

class AgentState(TypedDict):
    question: str              # User's input query
    public_ip: str | None      # From ipify tool
    latitude: float | None     # From ip_to_geo tool  
    longitude: float | None    # From ip_to_geo tool
    weather_data: str | None   # From weather_forecast tool
    answer: str | None         # Final LLM response
    messages: list             # Conversation history
    error: str | None          # Error tracking
    current_step: str          # Progress indicator

START
  ↓
┌───────────────────────┐
│  intent classification│ → [custom logic]
└──────┬────────────────┘
       ↓
┌─────────────┐
│  get_ip     │ → [ipify tool]
└──────┬──────┘
       │ [conditional: success/error]
       ↓
┌──────────────────┐
│ resolve_location │ → [ip_to_geo tool]
└──────┬───────────┘
       │ [conditional: success/error]
       ↓
┌──────────────┐
│fetch_weather │ → [weather_forecast tool]
└──────┬───────┘
       │ [conditional: success/error]
       ↓
┌────────────────┐
│generate_answer │ → [LLM synthesis]
└───────┬────────┘
        ↓
       END

     [error] ← Any failure routes here

def route_after_ip(state: AgentState) -> Literal["resolve_location", "error"]:
    """Route to next node or error based on IP fetch result"""
    if state.get("error") or not state.get("public_ip"):
        return "error"
    return "resolve_location"

python3 -m server.main

INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000

python3 -m agent.main

============================================================
Data Center Weather Agent (Custom StateGraph)
============================================================

Connected to MCP Server
LLM configured with fallback: Gemini -> LongCat

Graph structure built successfully

Agent Ready! (Type 'quit' to exit)
------------------------------------------------------------

Enter your question: What is the weather forecast of the data center?

============================================================
EXECUTION TRACE
============================================================

[Step 1: IP Discovery]
  Tool: ipify
  Result: 174.162.142.78

[Step 2: Location Resolution]
  Tool: ip_to_geo
  Input: 174.162.142.78
  Result: 40.3495, -111.8998

[Step 3: Weather Retrieval]
  Tool: weather_forecast
  Input: lat=40.3495, lon=-111.8998
  Result: Temperature: 0.2 C, Windspeed: 2.2 km/h

[Step 4: Answer Generation]
  Generated answer successfully

============================================================
FINAL ANSWER
============================================================

The data center, located at 40.3495°N, 111.8998°W, currently has a 
temperature of 0.2°C and a wind speed of 2.2 km/h.

Enter your question: Where is the data center located?

[... processes IP and location lookup ...]

FINAL ANSWER

The data center is located at coordinates 40.3495°N, 111.8998°W 
(approximately in Utah, United States).

Enter your question: quit

datacenter_weather_agent/
├── server/
│   ├── main.py              # MCP server with FastMCP
│   ├── tools.py             # Tool implementations
│   └── __init__.py
├── agent/
│   ├── main.py              # Custom StateGraph agent (~570 lines)
│   ├── client.py            # MCP client wrapper
│   └── __init__.py
├── README.md                # This file
├── ARCHITECTURE.md          # Deep technical documentation
├── .env.example             # API key template
├── .env                     # Your API keys (create this)
└── requirements.txt         # Python dependencies

async def get_ip_node(state, tools):
    try:
        # ... tool execution ...
        return success_state
    except Exception as e:
        logger.error(f"Error in get_ip_node: {e}")
        return error_state

gemini_llm = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
longcat_llm = ChatOpenAI(model="LongCat-Flash-Chat", ...)
llm = gemini_llm.with_fallbacks([longcat_llm])

[Step 1: IP Discovery]
  Tool: ipify
  Result: 174.162.142.78

from langgraph.checkpoint.sqlite import SqliteSaver

# In build_graph():
async with SqliteSaver.from_conn_string("./checkpoints.db") as memory:
    graph = workflow.compile(checkpointer=memory)

# In main():
config = {"configurable": {"thread_id": "user-123"}}
result = await graph.ainvoke(state, config=config)

from langgraph.checkpoint.sqlite import SqliteSaver

# Compile with interrupt
graph = workflow.compile(
    checkpointer=memory,
    interrupt_before=["generate_answer"]  # Pause before final answer
)

# Resume after approval
graph.invoke(None, config=config)  # Continue from interrupt

weather_specialist = build_weather_subgraph()
location_specialist = build_location_subgraph()

workflow.add_node("weather", weather_specialist)
workflow.add_node("location", location_specialist)
workflow.add_conditional_edges("start", route_to_specialist, {...})

# Always run from project root:
cd /path/to/datacenter_weather_agent
python3 -m agent.main 

def route_after_ip(state: AgentState):
    if state.get("error"):
        return "error"
    
    # New: Check if IP is internal/private
    ip = state.get("public_ip", "")
    if ip.startswith("192.168.") or ip.startswith("10."):
        return "handle_private_ip"  # New node
    
    return "resolve_location"

import pytest
from agent.main import get_ip_node

@pytest.mark.asyncio
async def test_get_ip_node():
    mock_tools = [MockIPifyTool()]
    state = {"question": "test", "messages": []}
    
    result = await get_ip_node(state, mock_tools)
    
    assert result["public_ip"] is not None
    assert result["error"] is None

@pytest.mark.asyncio
async def test_full_workflow(mcp_server_running):
    graph = await build_graph(client, llm)
    initial_state = {...}
    
    final_state = await graph.ainvoke(initial_state)
    
    assert final_state["answer"] is not None
    assert "temperature" in final_state["answer"].lower()

export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY=your_key
export LANGCHAIN_PROJECT="datacenter-weather"