import asyncio
import json
import sys
import argparse
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.sse import sse_client
async def call_tool_stdio(tool_name: str, tool_args: dict):
"""
Call a tool on the MCP server using stdio communication.
Args:
tool_name: Name of the tool to call
tool_args: Dictionary of arguments for the tool
"""
# Define server parameters for stdio connection
server_params = StdioServerParameters(
command="python",
args=["server.py"],
env=None
)
print(f"Connecting to MCP server (stdio)...")
print(f"Tool: {tool_name}")
print(f"Arguments: {json.dumps(tool_args, indent=2)}\n")
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# List available tools
tools_list = await session.list_tools()
print("Available tools:")
for tool in tools_list.tools:
print(f" - {tool.name}: {tool.description}")
print()
# Call the requested tool
print(f"Calling {tool_name}...")
result = await session.call_tool(tool_name, arguments=tool_args)
print("\n--- MCP Inspector ---")
print(json.dumps(result.content, indent=2, default=str))
print("---------------------\n")
async def call_tool_http(tool_name: str, tool_args: dict, url: str):
"""
Call a tool on the MCP server using HTTP/SSE communication.
Args:
tool_name: Name of the tool to call
tool_args: Dictionary of arguments for the tool
url: Base URL of the MCP server
"""
print(f"Connecting to MCP server at {url}...")
print(f"Tool: {tool_name}")
print(f"Arguments: {json.dumps(tool_args, indent=2)}\n")
async with sse_client(url) as (read, write):
async with ClientSession(read, write) as session:
# Initialize the connection
await session.initialize()
# List available tools
tools_list = await session.list_tools()
print("Available tools:")
for tool in tools_list.tools:
print(f" - {tool.name}: {tool.description}")
print()
# Call the requested tool
print(f"Calling {tool_name}...")
result = await session.call_tool(tool_name, arguments=tool_args)
print("\n--- MCP Inspector ---")
print(json.dumps(result.content, indent=2, default=str))
print("---------------------\n")
def main():
"""
A simple MCP client to test the Confluence MCP server.
Usage:
python host.py [--http URL] <tool_name> '<json_args>'
Examples:
# Local stdio connection
python host.py search_confluence '{"query": "My search query"}'
python host.py read_page '{"page_id": "12345"}'
python host.py list_space_content '{"space_key": "TEAM"}'
# HTTP/SSE connection (e.g., to Docker container)
python host.py --http http://localhost:8000 search_confluence '{"query": "test"}'
"""
parser = argparse.ArgumentParser(
description="Confluence MCP Client",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=main.__doc__
)
parser.add_argument(
"--http",
metavar="URL",
help="Connect via HTTP/SSE to the specified URL (e.g., http://localhost:8000)"
)
parser.add_argument("tool_name", help="Name of the tool to call")
parser.add_argument("args", help="Tool arguments as JSON string")
args = parser.parse_args()
try:
tool_args = json.loads(args.args)
except json.JSONDecodeError:
print("Error: Invalid JSON arguments.")
parser.print_help()
sys.exit(1)
try:
if args.http:
asyncio.run(call_tool_http(args.tool_name, tool_args, args.http))
else:
asyncio.run(call_tool_stdio(args.tool_name, tool_args))
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()
