Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions bot/agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,12 +95,12 @@ def truncate_history(self, chat_id: int) -> None:
def from_dict(cls, name: str, config: dict[str, Any]) -> OpenAIAgent:
mcp_servers: list[MCPServerStreamableHttp | MCPServerStdio] = []
for mcp_srv in config.get("mcpServers", {}).values():
if "httpUrl" in mcp_srv:
if "url" in mcp_srv:
mcp_servers.append(
MCPServerStreamableHttp(
client_session_timeout_seconds=MCP_SESSION_TIMEOUT_SECONDS,
params={
"url": mcp_srv["httpUrl"],
"url": mcp_srv["url"],
"headers": mcp_srv.get("headers", {}),
},
)
Expand Down
53 changes: 53 additions & 0 deletions tests/test_agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
from unittest.mock import patch

import pytest
from agents.mcp import MCPServerStdio
from agents.mcp import MCPServerStreamableHttp
from agents.models.interface import Model
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.models.openai_responses import OpenAIResponsesModel
Expand Down Expand Up @@ -114,6 +116,57 @@ def test_from_dict_uses_default_without_instructions(self):
assert agent.agent.instructions == DEFAULT_INSTRUCTIONS


class TestFromDictMcpServers:
def test_url_creates_streamable_http_server(self):
config = {
"mcpServers": {
"my-server": {
"url": "http://localhost:8000/mcp",
}
},
}
agent = OpenAIAgent.from_dict("test", config)
assert len(agent.agent.mcp_servers) == 1
assert isinstance(agent.agent.mcp_servers[0], MCPServerStreamableHttp)

def test_url_passes_headers(self):
config = {
"mcpServers": {
"my-server": {
"url": "http://localhost:8000/mcp",
"headers": {"Authorization": "Bearer token"},
}
},
}
agent = OpenAIAgent.from_dict("test", config)
server = agent.agent.mcp_servers[0]
assert isinstance(server, MCPServerStreamableHttp)

def test_command_creates_stdio_server(self):
config = {
"mcpServers": {
"my-server": {
"command": "npx",
"args": ["-y", "some-mcp-server"],
}
},
}
agent = OpenAIAgent.from_dict("test", config)
assert len(agent.agent.mcp_servers) == 1
assert isinstance(agent.agent.mcp_servers[0], MCPServerStdio)

def test_mixed_servers(self):
config = {
"mcpServers": {
"remote": {"url": "http://localhost:8000/mcp"},
"local": {"command": "npx", "args": ["-y", "server"]},
},
}
agent = OpenAIAgent.from_dict("test", config)
types = {type(s) for s in agent.agent.mcp_servers}
assert types == {MCPServerStreamableHttp, MCPServerStdio}


class TestHistoryTruncation:
def test_default_max_turns(self):
assert MAX_TURNS == 10
Expand Down
Loading