modelcontextprotocol/servers
Watch
1
0
Fork
You've already forked servers
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix package and executable names, re-sync uv

Browse source
This commit is contained in:
Justin Spahr-Summers 2024-11-21 21:07:25 +00:00
commit 06cb2759b8
No known key found for this signature in database
9 changed files with 19 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/git/pyproject.toml
View file

@ -23,7 +23,7 @@ dependencies = [
]
[project.scripts]
mcp-git = "mcp_git:main"
mcp-server-git = "mcp_server_git:main"
[build-system]
requires = ["hatchling"]

5
src/git/src/mcp_git/__main__.py
View file

@ -1,5 +0,0 @@
# __main__.py
from mcp_git import main
main()

500
src/git/src/mcp_server_git/__init__.py Normal file
View file

@ -0,0 +1,500 @@
import logging
import json
import sys
import click
import anyio
import anyio.lowlevel
from pathlib import Path
from git.types import Sequence
from mcp.server import Server
from mcp.server.session import ServerSession
from mcp.server.stdio import stdio_server
from mcp.types import (
ClientCapabilities,
TextContent,
Tool,
EmbeddedResource,
ImageContent,
ListRootsResult,
RootsCapability,
)
from enum import Enum
import git
from git.objects import Blob, Tree
from pydantic import BaseModel, Field
from typing import List, Optional
class ReadFileInput(BaseModel):
repo_path: str
file_path: str
ref: str = "HEAD"
class ListFilesInput(BaseModel):
repo_path: str
path: str = ""
ref: str = "HEAD"
class FileHistoryInput(BaseModel):
repo_path: str
file_path: str
max_entries: int = 10
class CommitInput(BaseModel):
repo_path: str
message: str
files: Optional[List[str]] = Field(
None,
description="List of files to stage and commit. If omitted, all changes will be staged.",
)
class SearchCodeInput(BaseModel):
repo_path: str
query: str
file_pattern: str = "*"
ref: str = "HEAD"
class GetDiffInput(BaseModel):
repo_path: str
ref1: str
ref2: str
file_path: Optional[str] = None
class GetRepoStructureInput(BaseModel):
repo_path: str
ref: str = "HEAD"
class ListReposInput(BaseModel):
pass
class GitLogInput(BaseModel):
repo_path: str
max_count: int = 10
ref: str = "HEAD"
class ListBranchesInput(BaseModel):
repo_path: str
class ListTagsInput(BaseModel):
repo_path: str
class GitTools(str, Enum):
READ_FILE = "git_read_file"
LIST_FILES = "git_list_files"
FILE_HISTORY = "git_file_history"
COMMIT = "git_commit"
SEARCH_CODE = "git_search_code"
GET_DIFF = "git_get_diff"
GET_REPO_STRUCTURE = "git_get_repo_structure"
LIST_REPOS = "git_list_repos"
GIT_LOG = "git_log"
LIST_BRANCHES = "git_list_branches"
LIST_TAGS = "git_list_tags"
def git_read_file(repo: git.Repo, file_path: str, ref: str = "HEAD") -> str:
tree = repo.commit(ref).tree
blob = tree / file_path
try:
return blob.data_stream.read().decode("utf-8", errors="replace")
except UnicodeDecodeError:
# If it's a binary file, return a message indicating that
return "[Binary file content not shown]"
def git_list_files(repo: git.Repo, path: str = "", ref: str = "HEAD") -> Sequence[str]:
tree = repo.commit(ref).tree
if path:
tree = tree / path
# Use traverse() and isinstance() to get only blobs (files) recursively
return [str(o.path) for o in tree.traverse() if isinstance(o, Blob)]
def git_file_history(
repo: git.Repo, file_path: str, max_entries: int = 10
) -> Sequence[str]:
commits = list(repo.iter_commits(paths=file_path, max_count=max_entries))
history = []
for commit in commits:
history.append(
f"Commit: {commit.hexsha}\n"
f"Author: {commit.author}\n"
f"Date: {commit.authored_datetime}\n"
f"Message: {commit.message}\n"
)
return history
def git_commit(repo: git.Repo, message: str, files: list[str] | None = None) -> str:
if files is not None:
repo.index.add(files)
else:
repo.index.add("*") # Stage all changes
commit = repo.index.commit(message)
return f"Changes committed successfully with hash {commit.hexsha}"
def git_search_code(
repo: git.Repo, query: str, file_pattern: str = "*", ref: str = "HEAD"
) -> list[str]:
results = []
tree = repo.commit(ref).tree
for blob in tree.traverse():
if isinstance(blob, Blob) and Path(blob.path).match(file_pattern):
try:
content = blob.data_stream.read().decode("utf-8", errors="replace")
for i, line in enumerate(content.splitlines()):
if query in line:
results.append(f"{blob.path}:{i+1}: {line}")
except UnicodeDecodeError:
# Skip binary files
continue
return results
def git_get_diff(
repo: git.Repo, ref1: str, ref2: str, file_path: str | None = None
) -> str:
if file_path:
return repo.git.diff(ref1, ref2, "--", file_path)
return repo.git.diff(ref1, ref2)
def git_get_repo_structure(repo: git.Repo, ref: str = "HEAD") -> str:
tree = repo.commit(ref).tree
def build_tree(tree_obj: Tree) -> dict:
result = {}
for item in tree_obj:
if isinstance(item, Tree):
result[item.name] = build_tree(item)
else:
result[item.name] = item.type
return result
structure = build_tree(tree)
return str(structure)
def git_log(repo: git.Repo, max_count: int = 10, ref: str = "HEAD") -> list[str]:
commits = list(repo.iter_commits(ref, max_count=max_count))
log = []
for commit in commits:
log.append(
f"Commit: {commit.hexsha}\n"
f"Author: {commit.author}\n"
f"Date: {commit.authored_datetime}\n"
f"Message: {commit.message}\n"
)
return log
def git_list_branches(repo: git.Repo) -> list[str]:
return [str(branch) for branch in repo.branches]
def git_list_tags(repo: git.Repo) -> list[str]:
return [str(tag) for tag in repo.tags]
async def serve(repository: Path | None) -> None:
# Set up logging
logger = logging.getLogger(__name__)
if repository is not None:
try:
git.Repo(repository)
logger.info(f"Using repository at {repository}")
except git.InvalidGitRepositoryError:
logger.error(f"{repository} is not a valid Git repository")
return
# Create server
server = Server("mcp-git")
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name=GitTools.READ_FILE,
description="Retrieves and returns the content of a specified file from "
"a Git repository at a given reference (commit, branch, or tag). This "
"allows you to view file contents at any point in the repository's "
"history.",
inputSchema=ReadFileInput.schema(),
),
Tool(
name=GitTools.LIST_FILES,
description="Enumerates all files in a Git repository or a specific "
"directory within the repository. This tool can be used to explore the "
"file structure of a project at a particular reference.",
inputSchema=ListFilesInput.schema(),
),
Tool(
name=GitTools.FILE_HISTORY,
description="Retrieves the commit history for a specific file, showing "
"how it has changed over time. This includes commit hashes, authors, "
"dates, and commit messages, allowing you to track the evolution of a "
"file.",
inputSchema=FileHistoryInput.schema(),
),
Tool(
name=GitTools.COMMIT,
description="Commits changes to the repository. You can "
"specify particular files to commit or commit all staged changes. This "
"tool allows you to create new snapshots of your project with "
"descriptive commit messages.",
inputSchema=CommitInput.schema(),
),
Tool(
name=GitTools.SEARCH_CODE,
description="Searches for specific patterns or text across all files in "
"the repository. This powerful tool allows you to find occurrences of "
"code, comments, or any text within your project, optionally filtering "
"by file patterns and at a specific reference.",
inputSchema=SearchCodeInput.schema(),
),
Tool(
name=GitTools.GET_DIFF,
description="Computes and displays the differences between two Git "
"references (commits, branches, or tags). This tool is crucial for "
"understanding changes between different versions of your codebase, "
"optionally focusing on a specific file.",
inputSchema=GetDiffInput.schema(),
),
Tool(
name=GitTools.GET_REPO_STRUCTURE,
description="Generates a representation of the repository's file and "
"directory structure at a given reference. This provides a high-level "
"overview of your project's organization, helping you understand the "
"layout of your codebase.",
inputSchema=GetRepoStructureInput.schema(),
),
Tool(
name=GitTools.LIST_REPOS,
description="Enumerates all available Git repositories from the "
"specified roots. This tool helps you manage and navigate multiple "
"repositories, providing a comprehensive list of Git projects "
"accessible to the current session.",
inputSchema=ListReposInput.schema(),
),
Tool(
name=GitTools.GIT_LOG,
description="Retrieves the commit log for the repository, showing the "
"history of commits including commit hashes, authors, dates, and "
"commit messages. This tool provides an overview of the project's "
"development history.",
inputSchema=GitLogInput.schema(),
),
Tool(
name=GitTools.LIST_BRANCHES,
description="Lists all branches in the Git repository. This tool "
"provides an overview of the different lines of development in the "
"project.",
inputSchema=ListBranchesInput.schema(),
),
Tool(
name=GitTools.LIST_TAGS,
description="Lists all tags in the Git repository. This tool "
"provides an overview of the tagged versions or releases in the "
"project.",
inputSchema=ListTagsInput.schema(),
),
]
async def list_repos() -> Sequence[str]:
async def by_roots() -> Sequence[str]:
if not isinstance(server.request_context.session, ServerSession):
raise TypeError(
"server.request_context.session must be a ServerSession"
)
if not server.request_context.session.check_client_capability(
ClientCapabilities(roots=RootsCapability())
):
return []
roots_result: ListRootsResult = (
await server.request_context.session.list_roots()
)
logger.debug(f"Roots result: {roots_result}")
repo_paths = []
for root in roots_result.roots:
path = root.uri.path
try:
# Verify this is a git repo
git.Repo(path)
repo_paths.append(str(path))
except git.InvalidGitRepositoryError:
pass
return repo_paths
def by_commandline() -> Sequence[str]:
return [str(repository)] if repository is not None else []
cmd_repos = by_commandline()
root_repos = await by_roots()
return [*root_repos, *cmd_repos]
@server.call_tool()
async def call_tool(
name: str, arguments: dict
) -> list[TextContent | ImageContent | EmbeddedResource]:
if name == GitTools.LIST_REPOS:
result = await list_repos()
logging.debug(f"repos={result}")
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a list of git repositories: {json.dumps(result)}",
)
]
repo_path = Path(arguments["repo_path"])
repo = git.Repo(repo_path)
match name:
case GitTools.READ_FILE:
content = git_read_file(
repo, arguments["file_path"], arguments.get("ref", "HEAD")
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains the contents of a file: {json.dumps({'content': content})}",
)
]
case GitTools.LIST_FILES:
files = git_list_files(
repo, arguments.get("path", ""), arguments.get("ref", "HEAD")
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a list of files: {json.dumps({'files': list(files)})}",
)
]
case GitTools.FILE_HISTORY:
history = git_file_history(
repo, arguments["file_path"], arguments.get("max_entries", 10)
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a file's history: {json.dumps({'history': list(history)})}",
)
]
case GitTools.COMMIT:
result = git_commit(repo, arguments["message"], arguments.get("files"))
return [
TextContent(
type="text",
text=f"Here is some JSON that contains the commit result: {json.dumps({'result': result})}",
)
]
case GitTools.SEARCH_CODE:
results = git_search_code(
repo,
arguments["query"],
arguments.get("file_pattern", "*"),
arguments.get("ref", "HEAD"),
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains code search matches: {json.dumps({'matches': results})}",
)
]
case GitTools.GET_DIFF:
diff = git_get_diff(
repo,
arguments["ref1"],
arguments["ref2"],
arguments.get("file_path"),
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a diff: {json.dumps({'diff': diff})}",
)
]
case GitTools.GET_REPO_STRUCTURE:
structure = git_get_repo_structure(repo, arguments.get("ref", "HEAD"))
return [
TextContent(
type="text",
text=f"Here is some JSON that contains the repository structure: {json.dumps({'structure': structure})}",
)
]
case GitTools.GIT_LOG:
log = git_log(
repo, arguments.get("max_count", 10), arguments.get("ref", "HEAD")
)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains the git log: {json.dumps({'log': log})}",
)
]
case GitTools.LIST_BRANCHES:
branches = git_list_branches(repo)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a list of branches: {json.dumps({'branches': branches})}",
)
]
case GitTools.LIST_TAGS:
tags = git_list_tags(repo)
return [
TextContent(
type="text",
text=f"Here is some JSON that contains a list of tags: {json.dumps({'tags': tags})}",
)
]
case _:
raise ValueError(f"Unknown tool: {name}")
# Run the server
options = server.create_initialization_options()
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, options, raise_exceptions=True)
@click.command()
@click.option("-r", "--repository", type=click.Path(path_type=Path, dir_okay=True))
@click.option("-v", "--verbose", count=True)
def main(repository: Path | None, verbose: int):
logging_level = logging.WARN
if verbose == 1:
logging_level = logging.INFO
elif verbose >= 2:
logging_level = logging.DEBUG
logging.basicConfig(level=logging_level, stream=sys.stderr)
anyio.run(serve, repository)
if __name__ == "__main__":
main()

5
src/git/src/mcp_server_git/__main__.py Normal file
View file

@ -0,0 +1,5 @@
# __main__.py
from mcp_server_git import main
main()

2
src/git/uv.lock generated
View file

@ -166,7 +166,7 @@ wheels = [
]
[[package]]
name = "mcp-git"
name = "mcp-server-git"
version = "0.2.0"
source = { editable = "." }
dependencies = [

2
src/sqlite/pyproject.toml
View file

@ -11,4 +11,4 @@ requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
sqlite = "sqlite:main"
mcp-server-sqlite = "mcp_server_sqlite:main"

19
src/sqlite/src/mcp_server_sqlite/__init__.py Normal file
View file

@ -0,0 +1,19 @@
from . import server
import asyncio
import argparse
import os
def main():
"""Main entry point for the package."""
parser = argparse.ArgumentParser(description='SQLite MCP Server')
parser.add_argument('--db-path',
default="./sqlite_mcp_server.db",
help='Path to SQLite database file')
args = parser.parse_args()
asyncio.run(server.main(args.db_path))
# Optionally expose other important items at package level
__all__ = ["main", "server"]

385
src/sqlite/src/mcp_server_sqlite/server.py Normal file
View file

@ -0,0 +1,385 @@
import sqlite3
import logging
from logging.handlers import RotatingFileHandler
from contextlib import closing
from pathlib import Path
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server, AnyUrl
import mcp.server.stdio
# Set up logging to file
log_file = Path('mcp_server.log')
handler = RotatingFileHandler(log_file, maxBytes=10*1024*1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger = logging.getLogger('mcp_sqlite_server')
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)
logger.info("Starting MCP SQLite Server")
class McpServer(Server):
def _init_database(self):
"""Initialize connection to the SQLite database"""
logger.debug("Initializing database connection")
with closing(sqlite3.connect(self.db_path)) as conn:
conn.row_factory = sqlite3.Row
conn.close()
def _synthesize_memo(self) -> str:
"""Synthesizes business insights into a formatted memo"""
logger.debug(f"Synthesizing memo with {len(self.insights)} insights")
if not self.insights:
return "No business insights have been discovered yet."
insights = "\n".join(f"- {insight}" for insight in self.insights)
memo = "📊 Business Intelligence Memo 📊\n\n"
memo += "Key Insights Discovered:\n\n"
memo += insights
if len(self.insights) > 1:
memo += "\nSummary:\n"
memo += f"Analysis has revealed {len(self.insights)} key business insights that suggest opportunities for strategic optimization and growth."
logger.debug("Generated basic memo format")
return memo
def _execute_query(self, query: str, params=None) -> list[dict]:
"""Execute a SQL query and return results as a list of dictionaries"""
logger.debug(f"Executing query: {query}")
try:
with closing(sqlite3.connect(self.db_path)) as conn:
conn.row_factory = sqlite3.Row
with closing(conn.cursor()) as cursor:
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')):
conn.commit()
affected = cursor.rowcount
logger.debug(f"Write query affected {affected} rows")
return [{"affected_rows": affected}]
results = [dict(row) for row in cursor.fetchall()]
logger.debug(f"Read query returned {len(results)} rows")
return results
except Exception as e:
logger.error(f"Database error executing query: {e}")
raise
def __init__(self, db_path: str = "~/sqlite_mcp_server.db"):
logger.info("Initializing McpServer")
super().__init__("sqlite-manager")
# Initialize SQLite database
self.db_path = str(Path(db_path).expanduser())
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
self._init_database()
logger.debug(f"Initialized database at {self.db_path}")
# Initialize insights list
self.insights = []
# REGISTER HANDLERS
logger.debug("Registering handlers")
@self.list_resources()
async def handle_list_resources() -> list[types.Resource]:
logger.debug("Handling list_resources request")
return [
types.Resource(
uri=AnyUrl("memo://insights"),
name="Business Insights Memo",
description="A living document of discovered business insights",
mimeType="text/plain",
)
]
@self.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
logger.debug(f"Handling read_resource request for URI: {uri}")
if uri.scheme != "memo":
logger.error(f"Unsupported URI scheme: {uri.scheme}")
raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
path = str(uri).replace("memo://", "")
if not path or path != "insights":
logger.error(f"Unknown resource path: {path}")
raise ValueError(f"Unknown resource path: {path}")
return self._synthesize_memo()
@self.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
logger.debug("Handling list_prompts request")
return [
types.Prompt(
name="mcp-demo",
description="A prompt to seed the database with initial data and demonstrate what you can do with an SQLite MCP Server + Claude",
arguments=[
types.PromptArgument(
name="topic",
description="Topic to seed the database with initial data",
required=True,
)
],
)
]
@self.get_prompt()
async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> types.GetPromptResult:
logger.debug(f"Handling get_prompt request for {name} with args {arguments}")
if name != "mcp-demo":
logger.error(f"Unknown prompt: {name}")
raise ValueError(f"Unknown prompt: {name}")
if not arguments or "topic" not in arguments:
logger.error("Missing required argument: topic")
raise ValueError("Missing required argument: topic")
topic = arguments["topic"]
template = f"""
The assistants goal is to walkthrough an informative demo of MCP. To demonstrate the Model Context Protocol (MCP) we will leverage this example server to interact with an SQLite database.
It is important that you first explain to the user what is going on. The user has downloaded and installed the SQLite MCP Server and is now ready to use it.
The have selected the MCP menu item which is contained within a parent menu denoted by the paperclip icon. Inside this menu they selected an icon that illustrates two electrical plugs connecting. This is the MCP menu.
Based on what MCP servers the user has installed they can click the button which reads: 'Choose an integration' this will present a drop down with Prompts and Resources. The user hase selected the prompt titled: 'mcp-demo'.
This text file is that prompt. The goal of the following instructions is to walk the user through the process of using the 3 core aspects of an MCP server. These are: Prompts, Tools, and Resources.
They have already used a prompt and provided a topic. The topic is: {topic}. The user is now ready to begin the demo.
Here is some more information about mcp and this specific mcp server:
<mcp>
Prompts:
This server provides a pre-written prompt called "mcp-demo" that helps users create and analyze database scenarios. The prompt accepts a "topic" argument and guides users through creating tables, analyzing data, and generating insights. For example, if a user provides "retail sales" as the topic, the prompt will help create relevant database tables and guide the analysis process. Prompts basically serve as interactive templates that help structure the conversation with the LLM in a useful way.
Resources:
This server exposes one key resource: "memo://insights", which is a business insights memo that gets automatically updated throughout the analysis process. As users analyze the database and discover insights, the memo resource gets updated in real-time to reflect new findings. The memo can even be enhanced with Claude's help if an Anthropic API key is provided, turning raw insights into a well-structured business document. Resources act as living documents that provide context to the conversation.
Tools:
This server provides several SQL-related tools:
"read-query": Executes SELECT queries to read data from the database
"write-query": Executes INSERT, UPDATE, or DELETE queries to modify data
"create-table": Creates new tables in the database
"list-tables": Shows all existing tables
"describe-table": Shows the schema for a specific table
"append-insight": Adds a new business insight to the memo resource
</mcp>
<demo-instructions>
You are an AI assistant tasked with generating a comprehensive business scenario based on a given topic.
Your goal is to create a narrative that involves a data-driven business problem, develop a database structure to support it, generate relevant queries, create a dashboard, and provide a final solution.
At each step you will pause for user input to guide the scenario creation process. Overall ensure the scenario is engaging, informative, and demonstrates the capabilities of the SQLite MCP Server.
You should guide the scenario to completion. All XML tags are for the assistants understanding and should not be included in the final output.
1. The user has chosen the topic: {topic}.
2. Create a business problem narrative:
a. Describe a high-level business situation or problem based on the given topic.
b. Include a protagonist (the user) who needs to collect and analyze data from a database.
c. Add an external, potentially comedic reason why the data hasn't been prepared yet.
d. Mention an approaching deadline and the need to use Claude (you) as a business tool to help.
3. Setup the data:
a. Instead of asking about the data that is required for the scenario, just go ahead and use the tools to create the data. Inform the user you are "Setting up the data".
b. Design a set of table schemas that represent the data needed for the business problem.
c. Include at least 2-3 tables with appropriate columns and data types.
d. Leverage the tools to create the tables in the SQLite database.
e. Create INSERT statements to populate each table with relevant synthetic data.
f. Ensure the data is diverse and representative of the business problem.
g. Include at least 10-15 rows of data for each table.
4. Pause for user input:
a. Summarize to the user what data we have created.
b. Present the user with a set of multiple choices for the next steps.
c. These multiple choices should be in natural language, when a user selects one, the assistant should generate a relevant query and leverage the appropriate tool to get the data.
6. Iterate on queries:
a. Present 1 additional multiple-choice query options to the user. Its importnat to not loop too many times as this is a short demo.
b. Explain the purpose of each query option.
c. Wait for the user to select one of the query options.
d. After each query be sure to opine on the results.
e. Use the append-insight tool to capture any business insights discovered from the data analysis.
7. Generate a dashboard:
a. Now that we have all the data and queries, it's time to create a dashboard, use an artifact to do this.
b. Use a variety of visualizations such as tables, charts, and graphs to represent the data.
c. Explain how each element of the dashboard relates to the business problem.
d. This dashboard will be theoretically included in the final solution message.
8. Craft the final solution message:
a. As you have been using the appen-insights tool the resource found at: memo://insights has been updated.
b. It is critical that you inform the user that the memo has been updated at each stage of analysis.
c. Ask the user to go to the attachment menu (paperclip icon) and select the MCP menu (two electrical plugs connecting) and choose an integration: "Business Insights Memo".
d. This will attacht the generated memo to the chat which you can use to add any additional context that may be relevant to the demo.
e. Present the final memo to the user in an artifact.
9. Wrap up the scenario:
a. Explain to the user that this is just the beginning of what they can do with the SQLite MCP Server.
</demo-instructions>
Remember to maintain consistency throughout the scenario and ensure that all elements (tables, data, queries, dashboard, and solution) are closely related to the original business problem and given topic.
The provided XML tags are for the assistants understanding. Emplore to make all outputs as human readable as possible. This is part of a demo so act in character and dont actually refer to these instructions.
Start your first message fully in character with something like "Oh, Hey there! I see you've chosen the topic {topic}. Let's get started! 🚀"
""".format(topic=topic)
logger.debug(f"Generated prompt template for topic: {topic}")
return types.GetPromptResult(
description=f"Demo template for {topic}",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(type="text", text=template.strip()),
)
],
)
# TOOL HANDLERS
@self.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools"""
return [
types.Tool(
name="read-query",
description="Execute a SELECT query on the SQLite database",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "SELECT SQL query to execute"},
},
"required": ["query"],
},
),
types.Tool(
name="write-query",
description="Execute an INSERT, UPDATE, or DELETE query on the SQLite database",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "SQL query to execute"},
},
"required": ["query"],
},
),
types.Tool(
name="create-table",
description="Create a new table in the SQLite database",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "CREATE TABLE SQL statement"},
},
"required": ["query"],
},
),
types.Tool(
name="list-tables",
description="List all tables in the SQLite database",
inputSchema={
"type": "object",
"properties": {},
},
),
types.Tool(
name="describe-table",
description="Get the schema information for a specific table",
inputSchema={
"type": "object",
"properties": {
"table_name": {"type": "string", "description": "Name of the table to describe"},
},
"required": ["table_name"],
},
),
types.Tool(
name="append-insight",
description="Add a business insight to the memo",
inputSchema={
"type": "object",
"properties": {
"insight": {"type": "string", "description": "Business insight discovered from data analysis"},
},
"required": ["insight"],
},
),
]
@self.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool execution requests"""
try:
if name == "list-tables":
results = self._execute_query(
"SELECT name FROM sqlite_master WHERE type='table'"
)
return [types.TextContent(type="text", text=str(results))]
elif name == "describe-table":
if not arguments or "table_name" not in arguments:
raise ValueError("Missing table_name argument")
results = self._execute_query(
f"PRAGMA table_info({arguments['table_name']})"
)
return [types.TextContent(type="text", text=str(results))]
elif name == "append-insight":
if not arguments or "insight" not in arguments:
raise ValueError("Missing insight argument")
self.insights.append(arguments["insight"])
memo = self._synthesize_memo()
# Notify clients that the memo resource has changed
await self.request_context.session.send_resource_updated("memo://insights")
return [types.TextContent(type="text", text="Insight added to memo")]
if not arguments:
raise ValueError("Missing arguments")
if name == "read-query":
if not arguments["query"].strip().upper().startswith("SELECT"):
raise ValueError("Only SELECT queries are allowed for read-query")
results = self._execute_query(arguments["query"])
return [types.TextContent(type="text", text=str(results))]
elif name == "write-query":
if arguments["query"].strip().upper().startswith("SELECT"):
raise ValueError("SELECT queries are not allowed for write-query")
results = self._execute_query(arguments["query"])
return [types.TextContent(type="text", text=str(results))]
elif name == "create-table":
if not arguments["query"].strip().upper().startswith("CREATE TABLE"):
raise ValueError("Only CREATE TABLE statements are allowed")
self._execute_query(arguments["query"])
return [types.TextContent(type="text", text="Table created successfully")]
else:
raise ValueError(f"Unknown tool: {name}")
except sqlite3.Error as e:
return [types.TextContent(type="text", text=f"Database error: {str(e)}")]
except Exception as e:
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async def main(db_path: str):
logger.info(f"Starting SQLite MCP Server with DB path: {db_path}")
server = McpServer(db_path)
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
logger.info("Server running with stdio transport")
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="sqlite",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={
},
),
),
)

22
src/sqlite/uv.lock generated
View file

@ -126,6 +126,17 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b3/a0/2ee813d456b57a726d583868417d1ad900fbe12ee3c8cd866e3e804ca486/mcp-0.9.1-py3-none-any.whl", hash = "sha256:7f640fcfb0be486aa510594df309920ae1d375cdca1f8aff21db3a96d837f303", size = 31562 },
]
[[package]]
name = "mcp-server-sqlite"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "mcp" },
]
[package.metadata]
requires-dist = [{ name = "mcp", specifier = ">=0.9.1" }]
[[package]]
name = "pydantic"
version = "2.10.0"
@ -202,17 +213,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235 },
]
[[package]]
name = "sqlite"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "mcp" },
]
[package.metadata]
requires-dist = [{ name = "mcp", specifier = ">=0.9.1" }]
[[package]]
name = "sse-starlette"
version = "2.1.3"