BirdScopeAI / langgraph_agent /simple_demo.py
facemelter's picture
Initial commit to hf space for hackathon
ff0e97f verified
"""
Simple LangGraph agent demo connecting to Modal bird classifier MCP
"""
import asyncio
import os
from dotenv import load_dotenv
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
load_dotenv()
async def create_bird_agent():
"""Create and return a bird classification agent."""
# 1. Configure MCP client for Modal bird classifier
print("[STATUS]: Connecting to Modal MCP server...")
client = MultiServerMCPClient({
"bird_classifier": {
"transport": "streamable_http",
"url": os.getenv("MODAL_MCP_URL"),
"headers": {
"X-API-Key": os.getenv("BIRD_CLASSIFIER_API_KEY")
}
}
})
# 2. Get tools from MCP server
print("[STATUS]: Loading MCP tools...")
tools = await client.get_tools()
print(f"[LOADED]: {len(tools)} tools - {[t.name for t in tools]}")
# 3. Create agent with model and tools
print("[STATUS]: Creating LangGraph agent...")
agent = create_agent(
model=ChatOpenAI(
model="gpt-4o-mini",
temperature=0
),
tools=tools,
system_prompt="""
You are a helpful bird identification assistant.
When a user provides an image URL, use the classify_from_url tool to identify the bird species.
Always report the species name and confidence score in a friendly manner."""
)
return agent
async def main():
"""Run simple bird classification agent demo"""
# Create agent
agent = await create_bird_agent()
# 4. Test with a bird image URL
print("\n"+"="*70)
print("[STATUS]: Testing bird classification...")
print("="*70+"\n")
test_url = "https://images.unsplash.com/photo-1445820200644-69f87d946277?w=400&auto=format&fit=crop&q=60&ixlib=rb-4.1.0&ixid=M3wxMjA3fDB8MHxzZWFyY2h8MTV8fGJpcmR8ZW58MHx8MHx8fDA%3D"
result = await agent.ainvoke({
"messages": [{
"role": "user",
"content": f"What bird species is this? {test_url}"
}]
})
# 5. Print results
print("\n[AGENT RESPONSE]:")
print(result["messages"][-1].content)
print("\n[DEMO COMPLETE!]")
if __name__ == "__main__":
asyncio.run(main())