Class MyAgent

examples/xpander/coding_agent.py:39–102

class MyAgent:
    def __init__(self):
        logger.info("🚀 Initializing MyAgent...")

        # Load config
        config_path = Path(__file__).parent / "xpander_config.json"
        config = json.loads(config_path.read_text())

        # Get API keys
        xpander_key = config.get("api_key") or os.getenv("XPANDER_API_KEY")
        agent_id = config.get("agent_id") or os.getenv("XPANDER_AGENT_ID")
        openai_key = os.getenv("OPENAI_API_KEY")

        if not all([xpander_key, agent_id, openai_key]):
            raise ValueError("Missing required API keys")

        # Initialize
        self.openai = AsyncOpenAI(api_key=openai_key)
        xpander_client = XpanderClient(api_key=xpander_key)
        self.agent_backend: Agent = xpander_client.agents.get(agent_id=agent_id)
        self.agent_backend.select_llm_provider(LLMProvider.OPEN_AI)

        logger.info(f"Agent: {self.agent_backend.name}")
        logger.info(f"Tools: {len(self.agent_backend.tools)} available")
        logger.info("✅ Ready!")

    async def run(self, user_txt_input: str) -> dict:
        step = 0
        start_time = time.perf_counter()
        tokens = Tokens(worker=LLMTokens(0, 0, 0))
        try:
            while not self.agent_backend.is_finished():
                step += 1
                logger.info(f"Step {step} - Calling LLM...")
                response = await self.openai.chat.completions.create(
                    model="gpt-4.1",
                    messages=self.agent_backend.messages,
                    tools=self.agent_backend.get_tools(),
                    tool_choice=self.agent_backend.tool_choice,
                    temperature=0,
                )
                if hasattr(response, "usage"):
                    tokens.worker.prompt_tokens += response.usage.prompt_tokens
                    tokens.worker.completion_tokens += response.usage.completion_tokens
                    tokens.worker.total_tokens += response.usage.total_tokens

                self.agent_backend.add_messages(response.model_dump())
                self.agent_backend.report_execution_metrics(llm_tokens=tokens, ai_model="gpt-4.1")
                tool_calls = self.agent_backend.extract_tool_calls(response.model_dump())

                if tool_calls:
                    logger.info(f"Executing {len(tool_calls)} tools...")
                    tool_results = await asyncio.to_thread(self.agent_backend.run_tools, tool_calls)
                    for res in tool_results:
                        emoji = "✅" if res.is_success else "❌"
                        logger.info(f"Tool result: {emoji} {res.function_name}")

            duration = time.perf_counter() - start_time