Spaces:
Sleeping
Sleeping
File size: 4,236 Bytes
676582c 8294216 676582c 8294216 676582c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
"""
MCP Tools Registration
Registers all MCP tools with the global tool registry.
Each tool is registered with its contract definition (name, description, parameters).
"""
import json
import logging
from pathlib import Path
from .tool_registry import tool_registry
from .tools.add_task import add_task
from .tools.list_tasks import list_tasks
from .tools.complete_task import complete_task
from .tools.delete_task import delete_task
from .tools.update_task import update_task
logger = logging.getLogger(__name__)
def load_tool_contract(tool_name: str) -> dict:
"""
Load MCP tool contract in a deployment-safe way
(works on Windows, Docker, Linux, cloud)
"""
# Always resolve from project root (/app in Docker)
project_root = Path(__file__).resolve().parents[2]
# parents[2] -> /app/src/mcp/__init__.py → /app
contract_path = (
project_root
/ "specs"
/ "001-openai-agent-mcp-tools"
/ "contracts"
/ f"{tool_name}.json"
)
if not contract_path.exists():
raise FileNotFoundError(f"Contract file not found: {contract_path}")
with contract_path.open("r", encoding="utf-8") as f:
return json.load(f)
def register_all_tools():
"""
Register all MCP tools with the global tool registry.
This function should be called during application startup to ensure
all tools are available to the agent.
"""
logger.info("Registering MCP tools...")
# Register add_task tool
try:
add_task_contract = load_tool_contract("add_task")
tool_registry.register_tool(
name=add_task_contract["name"],
description=add_task_contract["description"],
parameters=add_task_contract["parameters"],
handler=add_task
)
logger.info("Registered tool: add_task")
except Exception as e:
logger.error(f"Failed to register add_task tool: {str(e)}")
raise
# Register list_tasks tool
try:
list_tasks_contract = load_tool_contract("list_tasks")
tool_registry.register_tool(
name=list_tasks_contract["name"],
description=list_tasks_contract["description"],
parameters=list_tasks_contract["parameters"],
handler=list_tasks
)
logger.info("Registered tool: list_tasks")
except Exception as e:
logger.error(f"Failed to register list_tasks tool: {str(e)}")
raise
# Register complete_task tool
try:
complete_task_contract = load_tool_contract("complete_task")
tool_registry.register_tool(
name=complete_task_contract["name"],
description=complete_task_contract["description"],
parameters=complete_task_contract["parameters"],
handler=complete_task
)
logger.info("Registered tool: complete_task")
except Exception as e:
logger.error(f"Failed to register complete_task tool: {str(e)}")
raise
# Register delete_task tool
try:
delete_task_contract = load_tool_contract("delete_task")
tool_registry.register_tool(
name=delete_task_contract["name"],
description=delete_task_contract["description"],
parameters=delete_task_contract["parameters"],
handler=delete_task
)
logger.info("Registered tool: delete_task")
except Exception as e:
logger.error(f"Failed to register delete_task tool: {str(e)}")
raise
# Register update_task tool
try:
update_task_contract = load_tool_contract("update_task")
tool_registry.register_tool(
name=update_task_contract["name"],
description=update_task_contract["description"],
parameters=update_task_contract["parameters"],
handler=update_task
)
logger.info("Registered tool: update_task")
except Exception as e:
logger.error(f"Failed to register update_task tool: {str(e)}")
raise
logger.info(f"Successfully registered {len(tool_registry.list_tools())} MCP tools")
# Export the global registry instance for use in other modules
__all__ = ["tool_registry", "register_all_tools"]
|