state-of-mika 0.1.0

MCP client for AI agent frameworks

State of Mika: The Universal MCP Client for AI

State of Mika is a Python SDK that provides a simple, unified interface to the Model Context Protocol (MCP) ecosystem. It enables AI applications to seamlessly discover, connect to, and use MCP servers without needing to understand the underlying protocol details.

💡 What is MCP? The Model Context Protocol is an open-source, standardized communication protocol for AI models to interact with tools and services. Think of it as a "USB for AI" - a universal way to connect language models to external capabilities.

📋 Table of Contents

• Installation
• Quickstart
• Features
• Architecture Overview
• Detailed Usage Guide
  • Registry Management
  • Connecting to Servers
  • Working with Tools
  • API Key Management
  • Framework Integration
• Available MCP Servers
• Troubleshooting
• Advanced Usage
• Contributing
• License

🔧 Installation

Basic Installation

pip install state-of-mika

With Framework Support

# For LangChain integration
pip install state-of-mika[langchain]

# For FastAPI integration
pip install state-of-mika[fastapi]

# For developers contributing to the project
pip install state-of-mika[dev]

System Requirements

• Python 3.8 or higher
• Node.js 14+ (for npm-based MCP servers)
• Internet connection (for remote servers and server installations)

🚀 Quickstart

Here's a basic example to get you started:

import asyncio
import os
from state_of_mika import Mika

async def main():
    # Initialize the SDK
    mika = Mika()
    
    # See what MCP servers are available
    registry = mika.registry
    all_servers = registry.get_all_servers()
    print(f"Available servers: {len(all_servers)}")
    
    # Connect to a specific MCP server
    brave_client = await mika.get_or_create_client("brave_search")
    
    # Set your API key (required for most services)
    os.environ["BRAVE_SEARCH_API_KEY"] = "your-api-key-here"
    # Or securely save it for future sessions:
    # await mika.registry.save_api_key("brave_search", "your-api-key")
    
    # List available tools
    tools = await brave_client.list_tools()
    print(f"Available tools: {[t['name'] for t in tools]}")
    
    # Call a tool
    search_results = await brave_client.call_tool(
        "brave_web_search", 
        {"query": "What is Model Context Protocol?"}
    )
    
    # Process the results
    print("\nSearch results:")
    for result in search_results.get("webPages", {}).get("value", []):
        print(f"- {result['name']}: {result['url']}")
    
    # Clean up
    await mika.disconnect_all()

# Run the example
asyncio.run(main())

📊 Example Use Cases

Here are some real-world examples of how to use State of Mika for common tasks:

Example 1: Multi-Service Research Assistant

This example shows how to combine multiple services to create a research assistant that searches the web, academic papers, and creates a summary:

import asyncio
import os
from state_of_mika import Mika

async def research_topic(topic: str, output_file: str = "research_results.md"):
    """Research a topic using multiple sources and create a markdown summary."""
    
    # Initialize Mika
    mika = Mika()
    
    # Set API keys
    os.environ["BRAVE_SEARCH_API_KEY"] = "your-brave-search-key"
    
    try:
        # Connect to multiple services
        brave = await mika.get_or_create_client("brave_search")
        arxiv = await mika.get_or_create_client("arxiv")
        fs = await mika.get_or_create_client("filesystem")
        
        # Get web search results
        print(f"Searching the web for: {topic}")
        web_results = await brave.call_tool("brave_web_search", {"query": topic, "count": 3})
        
        # Get academic papers
        print(f"Finding academic papers on: {topic}")
        papers = await arxiv.call_tool("arxiv_search", {"query": topic, "max_results": 2})
        
        # Create markdown summary
        content = f"# Research: {topic}\n\n## Web Results\n\n"
        
        for result in web_results.get("webPages", {}).get("value", []):
            content += f"### [{result['name']}]({result['url']})\n"
            content += f"{result['snippet']}\n\n"
        
        content += "\n## Academic Papers\n\n"
        
        for paper in papers.get("papers", []):
            content += f"### [{paper['title']}]({paper['url']})\n"
            content += f"**Authors**: {paper['authors']}\n"
            content += f"**Abstract**: {paper['summary']}\n\n"
        
        # Save to file
        print(f"Writing results to {output_file}")
        await fs.call_tool("write_file", {
            "path": output_file,
            "content": content
        })
        
        print(f"Research complete! Results saved to {output_file}")
        
    except Exception as e:
        print(f"Error during research: {e}")
    
    finally:
        # Clean up
        await mika.disconnect_all()

# Run the research assistant
asyncio.run(research_topic("Model Context Protocol for AI"))

Example 2: Database and Chart Generator

This example shows how to query a database, process the results, and create a visualization:

import asyncio
import os
import matplotlib.pyplot as plt
import pandas as pd
from state_of_mika import Mika

async def generate_sales_report():
    """Query a database and generate a sales report with chart."""
    
    # Initialize Mika
    mika = Mika()
    
    try:
        # Connect to PostgreSQL and filesystem servers
        postgres = await mika.get_or_create_client("postgres")
        fs = await mika.get_or_create_client("filesystem")
        
        # Set connection parameters through environment variables
        os.environ["POSTGRES_CONNECTION_STRING"] = "postgresql://user:password@localhost:5432/sales_db"
        
        # Query the database
        print("Querying sales data...")
        result = await postgres.call_tool("execute_query", {
            "query": """
                SELECT 
                    date_trunc('month', order_date) as month,
                    SUM(amount) as total_sales
                FROM sales
                WHERE order_date >= NOW() - INTERVAL '1 year'
                GROUP BY 1
                ORDER BY 1
            """
        })
        
        if not result or 'rows' not in result:
            raise ValueError("Database query returned no results")
        
        # Process the data with pandas
        print("Processing results...")
        df = pd.DataFrame(result['rows'])
        
        # Create a visualization
        print("Generating chart...")
        plt.figure(figsize=(10, 6))
        plt.plot(df['month'], df['total_sales'], marker='o')
        plt.title('Monthly Sales - Last 12 Months')
        plt.xlabel('Month')
        plt.ylabel('Total Sales ($)')
        plt.grid(True)
        plt.tight_layout()
        
        # Save the chart
        chart_path = 'monthly_sales_chart.png'
        plt.savefig(chart_path)
        
        # Create a report
        report_content = "# Monthly Sales Report\n\n"
        report_content += f"## Summary\n\n"
        report_content += f"Total Sales: ${df['total_sales'].sum():,.2f}\n"
        report_content += f"Average Monthly Sales: ${df['total_sales'].mean():,.2f}\n"
        report_content += f"Best Month: {df.loc[df['total_sales'].idxmax(), 'month']} (${df['total_sales'].max():,.2f})\n\n"
        report_content += f"![Monthly Sales Chart]({chart_path})\n"
        
        # Save the report
        await fs.call_tool("write_file", {
            "path": "sales_report.md",
            "content": report_content
        })
        
        print("Report generated: sales_report.md")
        
    except Exception as e:
        print(f"Error generating report: {e}")
        
    finally:
        # Clean up
        await mika.disconnect_all()

# Run the report generator
asyncio.run(generate_sales_report())

Example 3: GitHub Repository Analyzer with Error Handling

This example demonstrates proper error handling while working with the GitHub API:

import asyncio
import os
from state_of_mika import Mika

async def analyze_repository(repo_owner: str, repo_name: str):
    """Analyze a GitHub repository and print insights."""
    
    # Initialize Mika
    mika = Mika()
    
    # Connect to GitHub
    try:
        # Get GitHub client
        github = await mika.get_or_create_client("github")
        
        # Set GitHub token (required for API access)
        os.environ["GITHUB_TOKEN"] = "your-github-token"
        
        # Get repository info
        print(f"Fetching information for {repo_owner}/{repo_name}...")
        repo_info = await github.call_tool("get_repository", {
            "owner": repo_owner,
            "repo": repo_name
        })
        
        # Get contributors
        try:
            contributors = await github.call_tool("get_contributors", {
                "owner": repo_owner,
                "repo": repo_name
            })
        except Exception as e:
            print(f"Warning: Couldn't fetch contributors: {e}")
            contributors = {"contributors": []}
        
        # Get issues
        try:
            issues = await github.call_tool("get_issues", {
                "owner": repo_owner,
                "repo": repo_name,
                "state": "open",
                "per_page": 5
            })
        except Exception as e:
            print(f"Warning: Couldn't fetch issues: {e}")
            issues = {"issues": []}
        
        # Print repository analysis
        print("\n" + "="*50)
        print(f"Repository Analysis: {repo_owner}/{repo_name}")
        print("="*50)
        
        print(f"\nDescription: {repo_info.get('description', 'No description')}")
        print(f"Stars: {repo_info.get('stargazers_count', 0)}")
        print(f"Forks: {repo_info.get('forks_count', 0)}")
        print(f"Open Issues: {repo_info.get('open_issues_count', 0)}")
        print(f"Language: {repo_info.get('language', 'Not specified')}")
        
        print("\nTop Contributors:")
        for idx, contributor in enumerate(contributors.get("contributors", [])[:5]):
            print(f"  {idx+1}. {contributor.get('login')} ({contributor.get('contributions')} contributions)")
        
        print("\nRecent Open Issues:")
        for idx, issue in enumerate(issues.get("issues", [])[:5]):
            print(f"  {idx+1}. {issue.get('title')} (#{issue.get('number')})")
        
        print("\nAnalysis Complete!")
        
    except Exception as e:
        print(f"Error analyzing repository: {e}")
        if "API rate limit exceeded" in str(e):
            print("Suggestion: Check your GitHub token or wait for rate limit to reset")
        elif "Not Found" in str(e):
            print(f"Suggestion: Verify that '{repo_owner}/{repo_name}' exists and is spelled correctly")
        
    finally:
        # Clean up
        await mika.disconnect_all()

# Run the repository analyzer
asyncio.run(analyze_repository("modelcontextprotocol", "mcp"))

Example 4: AI Assistant with LangChain and Multiple Tools

This example creates a more advanced AI assistant using LangChain and multiple MCP servers:

import asyncio
import os
from langchain.llms import OpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
from state_of_mika import Mika

async def run_ai_assistant():
    """Create an AI assistant with multiple capabilities."""
    
    # Set up API keys
    os.environ["OPENAI_API_KEY"] = "your-openai-key"
    os.environ["BRAVE_SEARCH_API_KEY"] = "your-brave-search-key"
    os.environ["GITHUB_TOKEN"] = "your-github-token"
    os.environ["WEATHER_API_KEY"] = "your-weather-api-key"
    
    # Initialize Mika
    mika = Mika()
    
    try:
        # Create LangChain components
        llm = OpenAI(temperature=0.7)
        memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
        
        # Initialize agent
        agent = initialize_agent(
            tools=[],  # We'll add tools via Mika
            llm=llm,
            agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
            verbose=True,
            memory=memory
        )
        
        # Connect to MCP servers
        print("Connecting to MCP servers...")
        adapter = await mika.connect(
            agent,
            servers=[
                "brave_search",   # For web search
                "github",         # For code and repo information
                "weather",        # For weather forecasts
                "filesystem"      # For file operations
            ]
        )
        
        print("\n" + "="*50)
        print("AI Assistant is ready! (Type 'exit' to quit)")
        print("="*50 + "\n")
        
        # Interactive chat loop
        while True:
            # Get user input
            user_input = input("You: ")
            
            if user_input.lower() in ["exit", "quit", "bye"]:
                print("AI: Goodbye!")
                break
            
            # Process with the agent
            try:
                response = await agent.arun(input=user_input)
                print(f"AI: {response}")
            except Exception as e:
                print(f"AI: I encountered an error: {e}")
                print("AI: Let me try to continue our conversation.")
        
    except Exception as e:
        print(f"Error initializing AI assistant: {e}")
    
    finally:
        # Clean up
        print("Cleaning up...")
        if 'agent' in locals():
            await mika.disconnect(agent)

# Run the AI assistant
if __name__ == "__main__":
    asyncio.run(run_ai_assistant())

Example 5: Custom MCP Server with State of Mika

This example shows how to create your own custom MCP server and connect to it:

# First, create a file named custom_mcp_server.py:
"""
A simple custom MCP server that provides text transformation tools.
"""
from mcp.server import Server, Tool

# Create the server
server = Server(name="Text Transformer")

@server.tool("capitalize")
async def capitalize(text: str) -> dict:
    """
    Capitalize every word in the text.
    
    Args:
        text: The input text to capitalize
    
    Returns:
        A dictionary with the capitalized text
    """
    return {
        "original": text,
        "transformed": text.title(),
        "operation": "capitalize"
    }

@server.tool("reverse")
async def reverse(text: str) -> dict:
    """
    Reverse the input text.
    
    Args:
        text: The input text to reverse
    
    Returns:
        A dictionary with the reversed text
    """
    return {
        "original": text,
        "transformed": text[::-1],
        "operation": "reverse"
    }

# Start the server when run directly
if __name__ == "__main__":
    server.start()

# Then, in your main script:
import asyncio
from state_of_mika import Mika

async def use_custom_server():
    """Use our custom MCP server with State of Mika."""
    
    # Initialize Mika
    mika = Mika()
    
    try:
        # Connect to our custom server
        print("Connecting to custom MCP server...")
        client = await mika.connect_to_server("python custom_mcp_server.py")
        
        # List available tools
        tools = await client.list_tools()
        print("Available tools:")
        for tool in tools:
            print(f"- {tool['name']}: {tool['description']}")
        
        # Use the capitalize tool
        text = "hello world, this is a custom mcp server"
        result = await client.call_tool("capitalize", {"text": text})
        print(f"\nCapitalized: {result['transformed']}")
        
        # Use the reverse tool
        result = await client.call_tool("reverse", {"text": text})
        print(f"Reversed: {result['transformed']}")
        
    except Exception as e:
        print(f"Error: {e}")
    
    finally:
        # Clean up
        if 'client' in locals():
            await client.disconnect()

# Run the example
asyncio.run(use_custom_server())

✨ Features

• Unified Interface: Access 140+ MCP servers through a single API
• Framework Integration: Works with popular AI frameworks like LangChain
• Server Management: Automatically discovers, installs, and manages MCP servers
• API Key Management: Securely stores and manages API keys
• Asynchronous Architecture: Built with asyncio for efficient request handling
• Tool Discovery: Automatically discovers available tools from connected servers
• Local and Remote Servers: Works with both local processes and remote API endpoints

🏗 Architecture Overview

State of Mika is organized into several key components:

1. Mika Core: The main entry point for the SDK
2. Registry: Manages available MCP servers and their metadata
3. Client: Handles communication with individual MCP servers
4. Transport: Manages connections to servers (stdio for local, HTTP for remote)
5. Adapters: Bridge between AI frameworks and the MCP ecosystem

📚 Detailed Usage Guide

Registry Management

The registry contains information about all available MCP servers.

from state_of_mika import Mika

# Initialize
mika = Mika()
registry = mika.registry

# List all available servers
all_servers = registry.get_all_servers()
for server in all_servers:
    print(f"- {server['name']}: {server['description']}")

# Find servers by tag
search_servers = [s for s in all_servers if "search" in s.get("tags", [])]
print(f"Found {len(search_servers)} search-related servers")

# Get details about a specific server
brave_server = registry.get_server("brave_search")
print(f"Brave Search: {brave_server['description']}")

# Add a custom server to the registry
await mika.add_server({
    "id": "my_custom_server",
    "name": "My Custom MCP Server",
    "description": "A custom MCP server implementation",
    "type": "local",
    "command": "python",
    "args": ["path/to/my_server.py"],
    "tags": ["custom", "experimental"]
})

Connecting to Servers

You can connect to MCP servers in several ways:

import asyncio
from state_of_mika import Mika

async def main():
    mika = Mika()
    
    # Method 1: Connect using server ID from registry
    brave_client = await mika.get_or_create_client("brave_search")
    
    # Method 2: Connect directly to a server script
    custom_client = await mika.connect_to_server("./my_custom_server.py")
    
    # Method 3: Connect to a remote server
    remote_client = await mika.connect_to_server("https://example.com/mcp-server")
    
    # Method 4: Connect with a full server configuration
    manual_client = await mika.connect_to_server({
        "id": "manual-server",
        "name": "Manual Config Server",
        "description": "Manually configured server",
        "type": "local",
        "command": "python",
        "args": ["-m", "my_server_module"]
    })
    
    # Clean up when done
    await brave_client.disconnect()
    await custom_client.disconnect()
    await remote_client.disconnect()
    await manual_client.disconnect()

asyncio.run(main())

Working with Tools

Once connected, you can discover and use tools provided by the server:

import asyncio
from state_of_mika import Mika

async def main():
    mika = Mika()
    client = await mika.get_or_create_client("brave_search")
    
    # List all available tools
    tools = await client.list_tools()
    print(f"Available tools on Brave Search server:")
    for tool in tools:
        print(f"- {tool['name']}: {tool['description']}")
    
    # Call a tool
    search_results = await client.call_tool("brave_web_search", {
        "query": "State of Mika MCP",
        "count": 5
    })
    
    # Process results
    for result in search_results.get("webPages", {}).get("value", []):
        print(f"- {result['name']}")
        print(f"  URL: {result['url']}")
        print(f"  Snippet: {result['snippet']}")
        print()
    
    # Clean up
    await client.disconnect()

asyncio.run(main())

API Key Management

Most MCP servers require API keys for authentication. State of Mika provides several ways to manage them:

import asyncio
import os
from state_of_mika import Mika

async def main():
    mika = Mika()
    
    # Method 1: Using environment variables (simplest)
    os.environ["BRAVE_SEARCH_API_KEY"] = "your-brave-search-api-key"
    os.environ["GITHUB_TOKEN"] = "your-github-token"
    
    # Method 2: Using the registry's API key storage (more secure)
    # Keys are stored in ~/.mika/api_keys.json
    await mika.registry.save_api_key("brave_search", "your-brave-search-api-key")
    await mika.registry.save_api_key("github", "your-github-token")
    
    # Method 3: Passing keys directly when connecting
    client = await mika.connect_to_server({
        "id": "brave_custom",
        "name": "Brave Search (Custom)",
        "type": "local",
        "command": "npx",
        "args": ["@modelcontextprotocol/server-brave-search"],
        "env": {
            "BRAVE_API_KEY": "your-brave-search-api-key"
        }
    })
    
    # Test the connection
    try:
        tools = await client.list_tools()
        print(f"Successfully connected with {len(tools)} tools available")
    except Exception as e:
        print(f"API key error: {e}")
    
    # Clean up
    await client.disconnect()

asyncio.run(main())

Obtaining API Keys

You need to obtain API keys from each service provider separately:

1. Brave Search:

   • Visit: https://brave.com/search/api/
   • Sign up for a developer account
   • Create a new API key
   • Set as BRAVE_SEARCH_API_KEY

2. GitHub:

   • Visit: https://github.com/settings/tokens
   • Generate a new token with appropriate scopes
   • Set as GITHUB_TOKEN

3. OpenWeather:

   • Visit: https://openweathermap.org/api
   • Sign up and get an API key
   • Set as OPENWEATHER_API_KEY

... and so on for other services.

Managing API Key Security

• Never hardcode API keys in your source code
• Use environment variables or the secure keystore for production
• Be aware of rate limits and billing for each API key
• Consider using a secrets manager for production deployments

Framework Integration

State of Mika integrates with AI agent frameworks like LangChain:

import asyncio
import os
from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
from state_of_mika import Mika

async def main():
    # Set up API keys
    os.environ["OPENAI_API_KEY"] = "your-openai-key"
    os.environ["BRAVE_SEARCH_API_KEY"] = "your-brave-search-key"
    os.environ["GITHUB_TOKEN"] = "your-github-token"
    
    # Create a LangChain agent
    llm = OpenAI(temperature=0)
    agent = initialize_agent(
        tools=[],  # Start with no tools, Mika will add them
        llm=llm,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True
    )
    
    # Connect the agent to MCP
    mika = Mika()
    
    # Specify which servers to use
    await mika.connect(
        agent,
        servers=["brave_search", "github", "filesystem"]
    )
    
    # Now the agent can use all tools from these servers
    result = await agent.arun(
        "Search for the top 3 MCP repositories on GitHub and create a file " +
        "called mcp_repos.txt with their names and descriptions."
    )
    
    print(result)
    
    # Disconnect when done
    await mika.disconnect(agent)

asyncio.run(main())

🌐 Available MCP Servers

State of Mika provides access to 140+ MCP servers. Here are some of the most popular:

Category	Server ID	Description	API Key Required?
Search	brave_search	Web search using Brave Search API	Yes
	google_search	Web search using Google Custom Search	Yes
	kagi	Kagi search engine API	Yes
	arXiv	Scientific paper search on arXiv	No
Development	github	GitHub repository management	Yes
	stackoverflow	StackOverflow Q&A search	No
	vscode	VS Code integration	No
	git	Git operations	No
Filesystem	filesystem	Local filesystem operations	No
	google_drive	Google Drive integration	Yes
	dropbox	Dropbox integration	Yes
Databases	postgres	PostgreSQL database client	Yes
	sqlite	SQLite database operations	No
	mongodb	MongoDB database client	Yes
AI/ML	openai	OpenAI API integration	Yes
	huggingface	Hugging Face model integration	Yes
	anthropic	Anthropic Claude integration	Yes
Utilities	weather	Weather data and forecasts	Yes
	calendar	Calendar management	Yes
	email	Email client	Yes

📋 For a complete list of available servers and their capabilities, run:

from state_of_mika import Mika
mika = Mika()
all_servers = mika.registry.get_all_servers()
for server in all_servers:
    print(f"{server['id']}: {server['description']}")

❓ Troubleshooting

Common Issues

API Key Errors

RuntimeError: Brave Search API key not found. Please set the BRAVE_SEARCH_API_KEY environment variable.

Solution: Set the required API key:

import os
os.environ["BRAVE_SEARCH_API_KEY"] = "your-api-key-here"

Server Installation Failures

Error installing npm package: @modelcontextprotocol/server-brave-search

Solution: Ensure Node.js and npm are installed and in your PATH.

# Check Node.js installation
node --version
npm --version

# Install manually if needed
npm install -g @modelcontextprotocol/server-brave-search

Connection Errors

Error connecting to MCP server: brave_search

Solution: Check network connectivity and firewall settings. For local servers, ensure you have the necessary permissions to run processes.

Tool Execution Timeouts

TimeoutError: Request brave_web_search timed out

Solution: Increase timeout settings or check if the server is overloaded.

# Increase timeout (if using a custom client implementation)
client = MCPClient(server_info, timeout=60)  # 60 seconds

Debugging

For detailed debugging, increase the logging level:

import logging
from state_of_mika import Mika

# Set up verbose logging
logging.basicConfig(level=logging.DEBUG)

# Initialize Mika
mika = Mika(log_level=logging.DEBUG)