import asyncio
import json
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import AsyncOpenAI
client = AsyncOpenAI(base_url="https://api.hcompany.ai/v1/", api_key=os.environ["HAI_API_KEY"])
MODEL = "holo4-35b-a3b"
ANSWER_TOOL = {
"type": "function",
"function": {
"name": "answer",
"description": "Provide a final answer",
"parameters": {
"type": "object",
"properties": {"content": {"type": "string", "description": "The answer content"}},
"required": ["content"],
},
},
}
def to_openai_tool(tool) -> dict:
return {
"type": "function",
"function": {"name": tool.name, "description": tool.description or "", "parameters": tool.input_schema},
}
async def run(task: str, max_steps: int = 10) -> str:
server = StdioServerParameters(command="uvx", args=["mcp-server-fetch"])
async with stdio_client(server) as (read, write), ClientSession(read, write) as session:
await session.initialize()
mcp_tools = (await session.list_tools()).tools
tools = [to_openai_tool(t) for t in mcp_tools] + [ANSWER_TOOL]
messages = [
{"role": "system", "content": "You are a research agent. Use your tools to gather facts, then answer."},
{"role": "user", "content": task},
]
for _ in range(max_steps):
resp = await client.chat.completions.create(
model=MODEL,
messages=messages,
tools=tools,
tool_choice="required",
temperature=0.8,
extra_body={"chat_template_kwargs": {"enable_thinking": True}},
)
msg = resp.choices[0].message
messages.append({"role": "assistant", "content": msg.content, "tool_calls": msg.tool_calls})
if not msg.tool_calls:
continue
call = msg.tool_calls[0]
args = json.loads(call.function.arguments)
if call.function.name == "answer":
return args["content"]
print(f"--- {call.function.name}({args})")
result = await session.call_tool(call.function.name, args)
text = "\n".join(c.text for c in result.content if c.type == "text")
messages.append({"role": "tool", "tool_call_id": call.id, "content": text[:8000]})
return "Step budget exhausted"
if __name__ == "__main__":
print(asyncio.run(run("Fetch https://docs.python.org/3/whatsnew/3.13.html and list the three headline features of Python 3.13.")))
