"""
OpenAI Agent that uses MCP server tools.
This agent:
1. Connects to MCP server
2. Automatically calls MCP tools based on user queries
3. Returns natural language responses
"""
import asyncio
from loguru import logger
# OpenAI Agents SDK
from openai_agents import Agent, MCPServer
from openai import AsyncAzureOpenAI
from src.settings import settings
class AppiumAgent:
"""AI Agent that connects to MCP server."""
def __init__(self):
"""Initialize agent with MCP connection."""
# ============================================================
# Step 1: Validate and Initialize OpenAI Client
# ============================================================
# Validate required settings
if not settings.AZURE_OPENAI_API_KEY:
raise ValueError("AZURE_OPENAI_API_KEY is required. Set it in .env file.")
if not settings.AZURE_OPENAI_ENDPOINT:
raise ValueError("AZURE_OPENAI_ENDPOINT is required. Set it in .env file.")
if not settings.AZURE_OPENAI_DEPLOYMENT:
raise ValueError(
"AZURE_OPENAI_DEPLOYMENT is required. Set it in .env file."
)
# Azure OpenAI
self.client = AsyncAzureOpenAI(
api_key=settings.AZURE_OPENAI_API_KEY,
api_version="2024-12-01-preview",
azure_endpoint=settings.AZURE_OPENAI_ENDPOINT,
)
self.model = settings.AZURE_OPENAI_DEPLOYMENT
# ============================================================
# Step 2: Connect to MCP Server
# ============================================================
self.mcp_server = MCPServer(
url=settings.MCP_SERVER_URL,
transport="sse", # Server-Sent Events
)
# ============================================================
# Step 3: Create Agent with MCP Server
# ============================================================
self.agent = Agent(
name="AppiumAgent",
instructions="You help users automate mobile app testing using available tools.",
client=self.client,
model=self.model,
mcp_servers=[self.mcp_server], # Connect MCP server here
)
logger.info("Agent initialized with MCP server connection")
async def query(self, user_message: str) -> str:
"""
Send query to agent and get response.
Agent will automatically call MCP tools as needed.
"""
logger.info(f"Query: {user_message}")
# Run agent with user message
response = await self.agent.run(
messages=[{"role": "user", "content": user_message}]
)
# Extract response text
return response.messages[-1].content
# ============================================================
# Usage Example
# ============================================================
async def main():
"""Call the agent to initialize the crawler and start automation."""
# Create agent
agent = AppiumAgent()
# Send queries
queries = [
"Initialize the crawler and start automation",
]
response = await agent.query(queries[0])
print(f"Agent: {response}")
if __name__ == "__main__":
asyncio.run(main())
