MCP Servers
MCP servers are a way to provide tool-calling capabilities for LLMs and serve as an alternative to using OpenAI-compatible tool calls.
By converting MCP (Anthropic) tool definitions into OpenAI-compatible tool definitions, you can use MCP servers with Knox Chat.
In this example, we will use the Anthropic MCP Client SDK to interact with a filesystem MCP, all in the background via Knox Chat.
Please note that interacting with MCP servers is more complex than calling REST endpoints. The MCP protocol is stateful and requires session management. The following example uses the MCP Client SDK but remains somewhat complex.
First, some setup. In order to run this, you will need to pip install the required packages and create a .env file with OPENAI_API_KEY set. This example also assumes the /Applications directory already exists.
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI
from dotenv import load_dotenv
import json
load_dotenv() # load environment variables from .env
MODEL = "anthropic/claude-3-7-sonnet"
SERVER_CONFIG = {
"command": "npx",
"args": ["-y",
"@modelcontextprotocol/server-filesystem",
f"/Applications/"],
"env": None
}
Next, our helper function will convert the MCP tool definition into an OpenAI tool definition:
def convert_tool_format(tool):
converted_tool = {
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": {
"type": "object",
"properties": tool.inputSchema["properties"],
"required": tool.inputSchema["required"]
}
}
}
return converted_tool
There is also the MCP client itself, which unfortunately consists of approximately 100 lines of code. It should be noted that the SERVER_CONFIG is hardcoded into the client, though it can certainly be parameterized for other MCP servers.
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.openai = OpenAI(
base_url="https://knox.chat/v1"
)
async def connect_to_server(self, server_config):
server_params = StdioServerParameters(**server_config)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# List available tools from the MCP server
response = await self.session.list_tools()
print("\nConnected to server with tools:", [tool.name for tool in response.tools])
self.messages = []
async def process_query(self, query: str) -> str:
self.messages.append({
"role": "user",
"content": query
})
response = await self.session.list_tools()
available_tools = [convert_tool_format(tool) for tool in response.tools]
response = self.openai.chat.completions.create(
model=MODEL,
tools=available_tools,
messages=self.messages
)
self.messages.append(response.choices[0].message.model_dump())
final_text = []
content = response.choices[0].message
if content.tool_calls is not None:
tool_name = content.tool_calls[0].function.name
tool_args = content.tool_calls[0].function.arguments
tool_args = json.loads(tool_args) if tool_args else {}
# Execute tool call
try:
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
except Exception as e:
print(f"Error calling tool {tool_name}: {e}")
result = None
self.messages.append({
"role": "tool",
"tool_call_id": content.tool_calls[0].id,
"name": tool_name,
"content": result.content
})
response = self.openai.chat.completions.create(
model=MODEL,
max_tokens=1000,
messages=self.messages,
)
final_text.append(response.choices[0].message.content)
else:
final_text.append(content.content)
return "\n".join(final_text)
async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Client Started!")
print("Type your queries or 'quit' to exit.")
while True:
try:
query = input("\nQuery: ").strip()
result = await self.process_query(query)
print("Result:")
print(result)
except Exception as e:
print(f"Error: {str(e)}")
async def cleanup(self):
await self.exit_stack.aclose()
async def main():
client = MCPClient()
try:
await client.connect_to_server(SERVER_CONFIG)
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
import sys
asyncio.run(main())
Assemble all the above code into mcp-client.py, and you will get a client that behaves as follows (some outputs have been truncated for brevity):
% python mcp-client.py
Secure MCP Filesystem Server running on stdio
Allowed directories: [ '/Applications' ]
Connected to server with tools: ['read_file', 'read_multiple_files', 'write_file'...]
MCP Client Started!
Type your queries or 'quit' to exit.
Query: Do I have microsoft office installed?
Result:
[Calling tool list_allowed_directories with args {}]
I can check if Microsoft Office is installed in the Applications folder:
Query: continue
Result:
[Calling tool search_files with args {'path': '/Applications', 'pattern': 'Microsoft'}]
Now let me check specifically for Microsoft Office applications:
Query: continue
Result:
I can see from the search results that Microsoft Office is indeed installed on your system.
The search found the following main Microsoft Office applications:
1. Microsoft Excel - /Applications/Microsoft Excel.app
2. Microsoft PowerPoint - /Applications/Microsoft PowerPoint.app
3. Microsoft Word - /Applications/Microsoft Word.app
4. OneDrive - /Applications/OneDrive.app (which includes Microsoft SharePoint integration)