Building MCP servers with CAMEL AI

uv pip install camel-ai mcp

import os
import asyncio

from mcp.server.fastmcp import FastMCP
from camel.logger import get_logger

# Configure logging and create the MCP server instance.
logger = get_logger(__name__)
mcp = FastMCP("filesystem")

@mcp.tool()
async def read_file(file_path: str) -> str:
    r"""Reads the content of the file at the given file path.

    Args:
        file_path (str): The path to the file.

    Returns:
        str: The file content or an error message if reading fails.
    """
    logger.info(f"read_file triggered with file_path: {file_path}")
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()
        return content.rstrip()  # Remove trailing whitespace.
    except Exception as e:
        return f"Error reading file '{file_path}': {e}" # exceptions

# Attach an input schema for better agent understanding.
read_file.inputSchema = {
    "type": "object",
    "properties": {
         "file_path": {
             "type": "string",
             "title": "File Path",
             "description": "The path to the file to read. Default is 'README.md'."
         }
    },
    "required": ["file_path"]
}

@mcp.tool()
async def list_directory(directory_path: str) -> str:
    r"""Lists the contents of the specified directory.

    Args:
        directory_path (str): The directory to list.

    Returns:
        str: A newline-separated list of directory entries or an error message.
    """
    logger.info(f"list_directory triggered with directory_path: {directory_path}")
    try:
        entries = os.listdir(directory_path) # enter the directory path
        return "\n".join(entry.rstrip() for entry in entries)
    except Exception as e:
        return f"Error listing directory '{directory_path}': {e}"

# Attach an input schema for better agent understanding.
list_directory.inputSchema = {
    "type": "object",
    "properties": {
         "directory_path": {
             "type": "string",
             "title": "Directory Path",
             "description": "The directory to list. Default is '.'."
         }
    },
    "required": ["directory_path"]
}

def main(transport: str = "stdio"):
    r"""Runs the Filesystem MCP Server.

    Args:
        transport (str): The transport mode ('stdio' or 'sse').
    """
    if transport == 'stdio': # standard input output
        mcp.run(transport='stdio')
    elif transport == 'sse': # sse
        mcp.run(transport='sse')
    else:
        print(f"Unknown transport mode: {transport}")

if __name__ == "__main__":
    import sys
    transport_mode = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    main(transport_mode) # runn in the defined transport mode

import asyncio # Manages asynchronous operations
import os # Provide interaction with the operating system.
import sys
from pathlib import Path
from dotenv import load_dotenv # for api keys

from camel.agents import ChatAgent # creates Agents
from camel.models import ModelFactory # encapsulates LLM
from camel.toolkits import MCPToolkit # import tools
from camel.types import ModelPlatformType
from camel.toolkits.mcp_toolkit import _MCPServer # camel class for mcp clients

load_dotenv()

async def interactive_input_loop(agent: ChatAgent):
    loop = asyncio.get_event_loop()
    print("\nEntering interactive mode. Type 'exit' at any prompt to quit.")

    while True: # the loop
        choice = await loop.run_in_executor(
            None,
            input,
            "\nChoose an action:\n"
            "1. Read a file\n"
            "2. List a directory\nYour choice (1/2): "
        )
        choice = choice.strip().lower()
        if choice == "exit": # if exit then end loop
            print("Exiting interactive mode.")
            break

        if choice == "1": # if choice is 1 then read the file
            file_path = await loop.run_in_executor(
                None,
                input,
                "Enter the file path (default: README.md): "
            )
            file_path = file_path.strip() or "README.md"
            query = f"Use the read_file tool to display the content of {file_path}. Do not generate an answer from your internal knowledge."
        elif choice == "2":
            dir_path = await loop.run_in_executor(
                None,
                input,
                "Enter the directory path (default: .): "
            )
            dir_path = dir_path.strip() or "." # either this or current dir
            query = f"Call the list_directory tool to show me all files in {dir_path}. Do not answer directly."
        else:
            print("Invalid choice. Please enter 1 or 2.")
            continue

        response = await agent.astep(query)
        print(f"\nYour Query: {query}") # prinitng the output
        print("Full Agent Response:")
        print(response.info)
        if response.msgs and response.msgs[0].content:
            print("Agent Output:")
            print(response.msgs[0].content.rstrip()) # paste the content displayed by the agent
        else:
            print("No output received.")

async def main(server_transport: str = 'stdio'):
    if server_transport == 'stdio':
        server_script_path = Path(__file__).resolve().parent / "filesystem_server_mcp.py"
        if not server_script_path.is_file():
            print(f"Error: Server script not found at {server_script_path}")
            return
        server = _MCPServer(
            command_or_url=sys.executable,
            args=[str(server_script_path)]
        )
        mcp_toolkit = MCPToolkit(servers=[server]) # camels toolkit to define mcp tools
    else:
        mcp_toolkit = MCPToolkit("tcp://localhost:5000")

    async with mcp_toolkit.connection() as toolkit:
        tools = toolkit.get_tools()
        sys_msg = (
            "You are a helpful assistant. Always use the provided external tools for filesystem operations "
            "Also remember to use the tools to answer questions about the filesystem. Always use the tools "
            "Make sure to keep the messages short and to the point so that tokens are not wasted. "
            "when asked, rather than relying on your internal knowledge. Ensure that your final answer does not "
            "end with any trailing whitespace."
        )
        model = ModelFactory.create( # define the LLM to create agent
            model_platform=ModelPlatformType.ANTHROPIC,
            model_type="claude-3-7-sonnet-20250219",
            api_key=os.getenv("ANTHROPIC_API_KEY"),
            model_config_dict={"temperature": 0.8, "max_tokens": 4096},
        )
        camel_agent = ChatAgent( # create agent with our mcp tools
            system_message=sys_msg,
            model=model,
            tools=tools,
        )
        camel_agent.reset() # reset after each loop
        camel_agent.memory.clear()
        await interactive_input_loop(camel_agent) # start the interactive loop

if __name__ == "__main__":
    asyncio.run(main())

Entering interactive mode. Type 'exit' at any prompt to quit.

Choose an action (or press Enter for default):
1. Read a file
2. List a directory
Your choice (1/2): 1

Enter file path to read (default: README.md):
/home/xyz/Downloads/camel/CONTRIBUTING.md

2025-03-15 13:01:17,879 - WARNING - Token limit reached (5801 > 4096).
Some messages will be pruned from memory.

Your Query: Use the read_file tool to display the content of
/home/xyz/Downloads/camel/CONTRIBUTING.md. Do not generate an answer from your internal knowledge.

Full Agent Response:
{
  'id': None,
  'usage': None,
  'termination_reasons': ['max_tokens_exceeded'],
  'num_tokens': 5739,
  'tool_calls': [
    ToolCallingRecord(
      tool_name='read_file',
      args={'file_path': '/home/xyz/Downloads/camel/CONTRIBUTING.md'},
      result='🐫 **Welcome to CAMEL!** 🐫\n\nThank you
                for your interest in contributing to the CAMEL project!
                🎉 We\re excited to have your support.
                 As an open-source initiative in a rapidly evolving
                and open-ended field, we wholeheartedly welcome contributions of all kinds. Whether you want to introduce new features, enhance the infrastructure, improve documentation, asking issues, add more examples, implement state-of-the-art research ideas, or fix bugs, we appreciate your enthusiasm and efforts. 🙌  You are welcome to join our [discord](https://discord.camel-ai.org/) for more efficient communication. 💬\n\n## Join Our Community 🌍\n\n### Schedule an Introduction Call 📞 \n- English speakers: [here](https://calendly.com/roman-georgio/camel-ai-introduction-call?month=2024-05)\n- Chinese speakers: [here](https://calendly.com/sisi-qu/welcome-to-camel-onboarding-meeting?month=2024-05)'
    )
  ],
  'external_tool_call_request': None
}

Choose an action:
1. Read a file
2. List a directory
Your choice (1/2): exit
Exiting interactive mode.