OmniXy

Universal Model Context Protocol (MCP) Client in python

OmniXy is a universal Model Context Protocol (MCP) client designed to enable seamless integration with any Large Language Model (LLM). It supports structured reasoning (chain-of-thought) and robust management of prompts, responses, and contexts according to the MCP specification.

Key Features

• Vendor-Agnostic: Easily switch between or integrate multiple LLM backends.
• Standards-Based: Leverages the Model Context Protocol to ensure consistency in request/response structures.
• Extensible: New LLM providers can be added via a driver pattern without refactoring the core.
• Reasoning-Capable: Orchestrate multi-step chain-of-thought processing for complex tasks.
• Secure & Scalable: Incorporates best practices for authentication, concurrency, caching, and data handling.
• Type-Safe: Uses Pydantic for data validation, serialization, and documentation.

Supported Models Names

For easier integration - our model name is structured as follows: provider-model-name:version.

as you can see in the models listed in the Open Router model list, as of February 25, 2025. For the most up-to-date list of supported models, refer to our open_router_models.json file.

Getting Started

1. Installation:

   # Using uv (Python)
   uv venv
   source .venv/bin/activate
   uv pip install -r requirements.txt

2. Configuration:

   • Set up your LLM provider credentials (e.g., Anthropic API key) in a secure vault or environment variables.

3. Usage Examples:

Basic Completion

    from universal_mcp_client import MCPClient

    client = MCPClient(provider="anthropic", api_key="YOUR_API_KEY")
    response = client.complete(
        messages=[{"role": "user", "content": "Explain quantum entanglement"}],
        model="claude-3.7-sonnet",
        stream=True
    )

    for chunk in response:
        print(chunk)
    ```

    ### Chain of Thought Reasoning

    ```python
    # Multi-step reasoning for complex tasks
    from universal_mcp_client import MCPClient, ChainOfThought

    client = MCPClient(provider="anthropic")
    chain = ChainOfThought(client)
    
    result = chain.solve([
        {"task": "Research phase", "prompt": "Analyze the given data"}, 
        {"task": "Planning", "prompt": "Create an action plan"}, 
        {"task": "Execution", "prompt": "Implement the solution"}
    ])
    ```

    ### Multiple Provider Integration

    ```python
    # Using multiple LLM providers in parallel
    from universal_mcp_client import MCPClient

    anthropic_client = MCPClient(provider="anthropic")
    cohere_client = MCPClient(provider="cohere")
    local_client = MCPClient(provider="local", model_path="/path/to/model")

    # Run inference across different providers
    responses = await asyncio.gather(
        anthropic_client.complete_async(messages=[{"role": "user", "content": prompt}]),
        cohere_client.complete_async(messages=[{"role": "user", "content": prompt}]),
        local_client.complete_async(messages=[{"role": "user", "content": prompt}])
    )
    ```

    ### Custom Provider Configuration

    ```python
    # Configure provider-specific parameters
    client = MCPClient(
        provider="anthropic",
        config={
            "temperature": 0.7,
            "max_tokens": 1000,
            "top_p": 0.95,
            "cache_enabled": True,
            "retry_strategy": "exponential",
            "timeout": 30
        }
    )
    ```

    ### Error Handling

    ```python
    from universal_mcp_client import MCPClient, MCPError

    client = MCPClient(provider="anthropic")
    
    try:
        response = client.complete(
            messages=[{"role": "user", "content": "Generate a story"}],
            model="claude-3.7-sonnet"
        )
    except MCPError as e:
        if e.error_type == "RateLimitError":
            print("Rate limit exceeded, retrying after cooldown")
        elif e.error_type == "AuthenticationError":
            print("Please check your API credentials")
        else:
            print(f"An error occurred: {e}")
    ```

### Using Pydantic Models

OmniXy uses Pydantic throughout the codebase for data validation, serialization, and documentation. You can leverage these models in your own code:

```python
from src.core.request import MCPRequest, Message
from src.core.response import MCPResponse
from src.core.settings import Settings, get_settings

# Create a request using Pydantic models
request = MCPRequest(
    provider="anthropic",
    model="claude-3.7-sonnet",
    messages=[
        Message(role="system", content="You are a helpful assistant."),
        Message(role="user", content="Tell me about quantum computing.")
    ]
)

# Access settings with type hints and validation
settings = get_settings()
print(f"Using default model: {settings.default_models['anthropic']}")

# Create your own models that extend OmniXy's models
from pydantic import BaseModel, Field
from typing import Dict, Any, Optional

class CustomCompletionRequest(BaseModel):
    request: MCPRequest
    metadata: Dict[str, Any] = Field(default_factory=dict)
    tracking_id: Optional[str] = None

Environment Variables and Settings

OmniXy uses Pydantic's BaseSettings to manage configuration through environment variables:

1. Copy the .env.example file to .env and fill in your API keys:

   cp .env.example .env
   # Edit .env with your API keys and settings

2. Settings are automatically loaded from the .env file:

from src.core.settings import get_settings

settings = get_settings()

# Access API keys and other settings
anthropic_api_key = settings.api.anthropic_api_key.get_secret_value()
default_timeout = settings.server.default_timeout

Schema Generation

OmniXy includes a schema generator that creates OpenAPI/JSON Schema documentation for all Pydantic models:

from src.core.schema_generator import generate_all_schemas
from pathlib import Path

# Generate schemas for all models
output_dir = Path("docs/schemas")
schemas = generate_all_schemas(output_dir)

Project Structure

OmniXy/
├── README.md
├── docs/
│   ├── architecture.md
│   ├── mcp_protocol.md
│   ├── provider_drivers.md
│   ├── orchestration.md
│   ├── security.md
├── src/
│   ├── __init__.py
│   ├── client.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── mcp_layer.py
│   │   ├── request.py
│   │   ├── response.py
│   │   ├── settings.py
│   │   ├── schema_generator.py
│   ├── drivers/
│   │   ├── __init__.py
│   │   ├── driver_factory.py
│   │   ├── anthropic_driver.py
│   │   ├── openai_driver.py
│   │   ├── cohere_driver.py
│   │   ├── gemini_driver.py
│   │   ├── local_llm_driver.py
│   ├── orchestration/
│   │   ├── __init__.py
│   │   ├── chain_of_thought.py
│   ├── workflow/
│   │   ├── __init__.py
│   │   ├── workflow_manager.py
│   ├── plugin/
│   │   ├── __init__.py
│   │   ├── plugin_manager.py
├── tests/
│   ├── unit/
│   ├── integration/

Testing

OmniXy maintains a comprehensive test suite with both unit and integration tests. We strive for 80% or higher code coverage.

Running Tests

# Install test dependencies
uv pip install pytest pytest-cov pytest-asyncio

# Run all tests
uv pip run pytest

# Run only unit tests
uv pip run pytest tests/unit

# Run only integration tests
uv pip run pytest tests/integration

# Run with coverage report
uv pip run pytest --cov=src tests/ --cov-report=term-missing --cov-report=xml

Test Structure

• Unit Tests: Located in tests/unit/

  • test_mcp_layer.py: Tests for core MCP protocol functionality
  • test_request.py: Request handling and validation
  • test_response.py: Response processing and formatting
  • test_workflow_manager.py: Workflow creation and execution
  • test_plugin_manager.py: Plugin system functionality
  • test_chain_of_thought.py: Chain-of-thought orchestration

• Integration Tests: Located in tests/integration/

  • test_workflow_plugin_integration.py: Workflow and plugin interaction
  • test_mcp_providers.py: Provider-specific integration tests

Writing Tests

When contributing new features:

1. Add corresponding unit tests in tests/unit/
2. Add integration tests if the feature interacts with other components
3. Ensure tests are properly documented
4. Verify test coverage remains at or above 80%

Contributing

Development Setup

1. Fork and clone the repository
2. Set up the development environment:

# Create and activate virtual environment
uv venv
source .venv/bin/activate

# Install dependencies
uv pip install -r requirements.txt

# Install development dependencies
uv pip install pytest pytest-cov pytest-asyncio ruff black mypy

Development Workflow

1. Create a new branch for your feature:

   git checkout -b feature/your-feature-name

2. Make your changes and ensure all tests pass:

   # Run tests
   uv pip run pytest
   
   # Run linters
   ruff check .
   black .
   mypy src tests

3. Commit your changes using conventional commits:

   git commit -m "feat: add new feature"

4. Push your branch and create a Pull Request

Continuous Integration

When you open a Pull Request targeting the main branch, our CI workflow will automatically:

1. Run tests on multiple Python versions (3.9, 3.10, 3.11)
2. Generate and upload test coverage reports
3. Run linting checks (ruff, black, mypy)
4. Verify documentation updates

Ensure all CI checks pass before requesting a review.

License

For commercial use, please contact the Licensor via LinkedIn or via X @elkriefy.

For more information: LICENSE