┌─────────────────────────────────────────────────────────────┐
│  Client Projects (Custom Agents, LangGraph, CrewAI, IDEs)   │
└──────────────────────────────┬──────────────────────────────┘
                               │ SSE (http://<host>:8080/sse)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│                      Omega MCP Server                       │
│                   (MCPServer SSE Engine)                    │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ ToolRegistry Engine (Dynamic Scanner & Binder)        │  │
│  │  ├── SystemStatusTool  (subclass of BaseTool)         │  │
│  │  ├── CalculateSumTool  (subclass of BaseTool)         │  │
│  │  └── <NewTool>         (subclass of BaseTool)         │  │
│  └───────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

{
  "success": true,
  "data": {
    "result_key": "result_value"
  },
  "error": null,
  "summary": "Concise natural language summary of what occurred."
}

omega-mcp-server/
├── Dockerfile
├── docker-compose.yaml
├── pyproject.toml
├── .dockerignore
├── README.md
├── scripts/
│   └── mcp_client_test.py     # Verification client test script
└── src/
    └── mcp_server/
        ├── __init__.py
        ├── common.py          # Standard AgentResponse contract model
        ├── registry.py        # BaseTool ABC & ToolRegistry engine
        ├── server.py          # MCPServer bootstrap & adapter binding
        └── tools/             # Class-based tool modules
            ├── __init__.py
            ├── system.py      # SystemStatusTool (host inspection)
            └── math.py        # CalculateSumTool (arithmetic computation)

uv sync

uv run python -m mcp_server.server

uv run python -m scripts.mcp_client_test

docker compose up --build -d

docker compose logs -f

curl -N http://localhost:8080/sse

from typing import Any
from pydantic import BaseModel, Field
from mcp_server.common import AgentResponse
from mcp_server.registry import BaseTool


# 1. Define input parameters and descriptions
class CustomToolInput(BaseModel):
    query: str = Field(..., description="Input query or target argument")
    limit: int = Field(default=5, description="Maximum number of items to return")


# 2. Implement the tool class
class CustomTool(BaseTool):
    name = "custom_tool"
    description = "Detailed explanation of what the tool accomplishes."
    args_schema = CustomToolInput

    async def run(self, query: str, limit: int = 5) -> dict:
        try:
            # Core execution logic
            result_payload = {"echo": query, "count": limit}

            return AgentResponse.ok(
                data=result_payload,
                summary=f"Processed query '{query}' with limit {limit}."
            ).model_dump()

        except Exception as exc:
            return AgentResponse.fail(str(exc)).model_dump()

import asyncio
import json
import os
from typing import Any, Callable, Dict, List
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client

DEFAULT_SSE_URL = os.getenv("OMEGA_MCP_URL", "http://localhost:8080/sse")


class OmegaMCPClient:
    """Manages SSE connection and execution for remote MCP tools."""

    def __init__(self, sse_url: str = DEFAULT_SSE_URL):
        self.sse_url = sse_url

    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """Call remote tool over SSE and unwrap the AgentResponse envelope."""
        async with sse_client(self.sse_url) as (read_stream, write_stream):
            async with ClientSession(read_stream, write_stream) as session:
                await session.initialize()
                response = await session.call_tool(tool_name, arguments=arguments)
                raw_text = response.content[0].text

                try:
                    payload = json.loads(raw_text)
                    if not payload.get("success", False):
                        return f"Tool Error ({tool_name}): {payload.get('error')}"
                    return str(payload.get("data"))
                except (json.JSONDecodeError, AttributeError):
                    return raw_text

    async def list_tools(self) -> List[str]:
        """Query available tool names exposed by the server."""
        async with sse_client(self.sse_url) as (read_stream, write_stream):
            async with ClientSession(read_stream, write_stream) as session:
                await session.initialize()
                res = await session.list_tools()
                return [t.name for t in res.tools]


# Shared client instance
client = OmegaMCPClient()


def create_agent_tool(name: str, description: str = "") -> Callable:
    """Factory that produces a generic callable compatible with agent frameworks."""

    async def async_dispatch(**kwargs) -> str:
        return await client.execute_tool(name, kwargs)

    def dispatch(**kwargs) -> str:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop and loop.is_running():
            import nest_asyncio
            nest_asyncio.apply()
            return loop.run_until_complete(async_dispatch(**kwargs))
        else:
            return asyncio.run(async_dispatch(**kwargs))

    dispatch.__name__ = name
    dispatch.__doc__ = description or f"Executes the remote MCP tool '{name}'."
    return dispatch


# Map remote tools to standard Python callables
get_system_status = create_agent_tool(
    name="get_system_status",
    description="Retrieves current host CPU and memory usage statistics."
)

calculate_sum = create_agent_tool(
    name="calculate_sum",
    description="Calculates the sum of two numbers (a, b)."
)

import os
# Replace with your framework's imports (Google GenAI, LangChain, CrewAI, AutoGen, etc.)
from your_agent_framework import Agent, llm

# Import pre-configured callables from the MCP client adapter
from mcp_client import calculate_sum, get_system_status

generic_agent = Agent(
    name="GenericTaskAgent",
    description="Executes tasks using modular tools served over MCP.",
    model=llm,
    tools=[get_system_status, calculate_sum],
    instruction="""You are a task execution agent.
    When asked about system metrics or arithmetic operations, call the appropriate 
    registered tool and summarize the result cleanly for the user.
    """
)

if __name__ == "__main__":
    # Example invocation
    prompt = "Check current system resources and calculate 45 + 55."
    print(f"User Request: {prompt}\n")

    # Run agent according to your framework API:
    # response = generic_agent.run(prompt)
    # print(response)

docker network create agent-network

services:
  omega-mcp-server:
    # ...
    networks:
      - agent-network

networks:
  agent-network:
    external: true

OMEGA_MCP_URL=http://omega_mcp_server:8080/sse