Dexter’s Blog

Introduction MCP is an open protocol that standardizes how applications provide context to large language models (LLMs). Why MCP MCP Architecture graph TB subgraph "MCP Host (AI Application)" Client1["MCP Client 1"] Client2["MCP Client 2"] Client3["MCP Client 3"] end Server1["MCP Server 1<br/>(e.g., Sentry)"] Server2["MCP Server 2<br/>(e.g., Filesystem)"] Server3["MCP Server 3<br/>(e.g., Database)"] Client1 ---|"One-to-one<br/>connection"| Server1 Client2 ---|"One-to-one<br/>connection"| Server2 Client3 ---|"One-to-one<br/>connection"| Server3 style Client1 fill:#e1f5fe style Client2 fill:#e1f5fe style Client3 fill:#e1f5fe style Server1 fill:#f3e5f5 style Server2 fill:#f3e5f5 style Server3 fill:#f3e5f5 Creating an MCP Client and an MCP Server From Function Call to MCP Whole Picture Tools/Resources/Prompts Discovery and Invocation Client Source Code from dotenv import load_dotenv from anthropic import Anthropic from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from contextlib import AsyncExitStack import json import asyncio import nest_asyncio nest_asyncio.apply() load_dotenv() class MCP_ChatBot: def __init__(self): self.exit_stack = AsyncExitStack() self.anthropic = Anthropic() # Tools list required for Anthropic API self.available_tools = [] # Prompts list for quick display self.available_prompts = [] # Sessions dict maps tool/prompt names or resource URIs to MCP client sessions self.sessions = {} async def connect_to_server(self, server_name, server_config): try: server_params = StdioServerParameters(**server_config) stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params) ) read, write = stdio_transport session = await self.exit_stack.enter_async_context( ClientSession(read, write) ) await session.initialize() try: # List available tools response = await session.list_tools() for tool in response.tools: self.sessions[tool.name] = session self.available_tools.append({ "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema }) # List available prompts prompts_response = await session.list_prompts() if prompts_response and prompts_response.prompts: for prompt in prompts_response.prompts: self.sessions[prompt.name] = session self.available_prompts.append({ "name": prompt.name, "description": prompt.description, "arguments": prompt.arguments }) # List available resources resources_response = await session.list_resources() if resources_response and resources_response.resources: for resource in resources_response.resources: resource_uri = str(resource.uri) self.sessions[resource_uri] = session except Exception as e: print(f"Error {e}") except Exception as e: print(f"Error connecting to {server_name}: {e}") async def connect_to_servers(self): try: with open("server_config.json", "r") as file: data = json.load(file) servers = data.get("mcpServers", {}) for server_name, server_config in servers.items(): await self.connect_to_server(server_name, server_config) except Exception as e: print(f"Error loading server config: {e}") raise async def process_query(self, query): messages = [{'role':'user', 'content':query}] while True: response = self.anthropic.messages.create( max_tokens = 2024, model = 'claude-3-7-sonnet-20250219', tools = self.available_tools, messages = messages ) assistant_content = [] has_tool_use = False for content in response.content: if content.type == 'text': print(content.text) assistant_content.append(content) elif content.type == 'tool_use': has_tool_use = True assistant_content.append(content) messages.append({'role':'assistant', 'content':assistant_content}) # Get session and call tool session = self.sessions.get(content.name) if not session: print(f"Tool '{content.name}' not found.") break result = await session.call_tool(content.name, arguments=content.input) messages.append({ "role": "user", "content": [ { "type": "tool_result", "tool_use_id": content.id, "content": result.content } ] }) # Exit loop if no tool was used if not has_tool_use: break async def get_resource(self, resource_uri): session = self.sessions.get(resource_uri) # Fallback for papers URIs - try any papers resource session if not session and resource_uri.startswith("papers://"): for uri, sess in self.sessions.items(): if uri.startswith("papers://"): session = sess break if not session: print(f"Resource '{resource_uri}' not found.") return try: result = await session.read_resource(uri=resource_uri) if result and result.contents: print(f"\nResource: {resource_uri}") print("Content:") print(result.contents[0].text) else: print("No content available.") except Exception as e: print(f"Error: {e}") async def list_prompts(self): """List all available prompts.""" if not self.available_prompts: print("No prompts available.") return print("\nAvailable prompts:") for prompt in self.available_prompts: print(f"- {prompt['name']}: {prompt['description']}") if prompt['arguments']: print(f" Arguments:") for arg in prompt['arguments']: arg_name = arg.name if hasattr(arg, 'name') else arg.get('name', '') print(f" - {arg_name}") async def execute_prompt(self, prompt_name, args): """Execute a prompt with the given arguments.""" session = self.sessions.get(prompt_name) if not session: print(f"Prompt '{prompt_name}' not found.") return try: result = await session.get_prompt(prompt_name, arguments=args) if result and result.messages: prompt_content = result.messages[0].content # Extract text from content (handles different formats) if isinstance(prompt_content, str): text = prompt_content elif hasattr(prompt_content, 'text'): text = prompt_content.text else: # Handle list of content items text = " ".join(item.text if hasattr(item, 'text') else str(item) for item in prompt_content) print(f"\nExecuting prompt '{prompt_name}'...") await self.process_query(text) except Exception as e: print(f"Error: {e}") async def chat_loop(self): print("\nMCP Chatbot Started!") print("Type your queries or 'quit' to exit.") print("Use @folders to see available topics") print("Use @<topic> to search papers in that topic") print("Use /prompts to list available prompts") print("Use /prompt <name> <arg1=value1> to execute a prompt") while True: try: query = input("\nQuery: ").strip() if not query: continue if query.lower() == 'quit': break # Check for @resource syntax first if query.startswith('@'): # Remove @ sign topic = query[1:] if topic == "folders": resource_uri = "papers://folders" else: resource_uri = f"papers://{topic}" await self.get_resource(resource_uri) continue # Check for /command syntax if query.startswith('/'): parts = query.split() command = parts[0].lower() if command == '/prompts': await self.list_prompts() elif command == '/prompt': if len(parts) < 2: print("Usage: /prompt <name> <arg1=value1> <arg2=value2>") continue prompt_name = parts[1] args = {} # Parse arguments for arg in parts[2:]: if '=' in arg: key, value = arg.split('=', 1) args[key] = value await self.execute_prompt(prompt_name, args) else: print(f"Unknown command: {command}") continue await self.process_query(query) except Exception as e: print(f"\nError: {str(e)}") async def cleanup(self): await self.exit_stack.aclose() async def main(): chatbot = MCP_ChatBot() try: await chatbot.connect_to_servers() await chatbot.chat_loop() finally: await chatbot.cleanup() if __name__ == "__main__": asyncio.run(main()) SSE Remote Server Source Code import arxiv import json import os from typing import List from mcp.server.fastmcp import FastMCP PAPER_DIR = "papers" # Initialize FastMCP server mcp = FastMCP("research", port=8001) @mcp.tool() def search_papers(topic: str, max_results: int = 5) -> List[str]: """ Search for papers on arXiv based on a topic and store their information. Args: topic: The topic to search for max_results: Maximum number of results to retrieve (default: 5) Returns: List of paper IDs found in the search """ # Use arxiv to find the papers client = arxiv.Client() # Search for the most relevant articles matching the queried topic search = arxiv.Search( query = topic, max_results = max_results, sort_by = arxiv.SortCriterion.Relevance ) papers = client.results(search) # Create directory for this topic path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_")) os.makedirs(path, exist_ok=True) file_path = os.path.join(path, "papers_info.json") # Try to load existing papers info try: with open(file_path, "r") as json_file: papers_info = json.load(json_file) except (FileNotFoundError, json.JSONDecodeError): papers_info = {} # Process each paper and add to papers_info paper_ids = [] for paper in papers: paper_ids.append(paper.get_short_id()) paper_info = { 'title': paper.title, 'authors': [author.name for author in paper.authors], 'summary': paper.summary, 'pdf_url': paper.pdf_url, 'published': str(paper.published.date()) } papers_info[paper.get_short_id()] = paper_info # Save updated papers_info to json file with open(file_path, "w") as json_file: json.dump(papers_info, json_file, indent=2) print(f"Results are saved in: {file_path}") return paper_ids @mcp.tool() def extract_info(paper_id: str) -> str: """ Search for information about a specific paper across all topic directories. Args: paper_id: The ID of the paper to look for Returns: JSON string with paper information if found, error message if not found """ for item in os.listdir(PAPER_DIR): item_path = os.path.join(PAPER_DIR, item) if os.path.isdir(item_path): file_path = os.path.join(item_path, "papers_info.json") if os.path.isfile(file_path): try: with open(file_path, "r") as json_file: papers_info = json.load(json_file) if paper_id in papers_info: return json.dumps(papers_info[paper_id], indent=2) except (FileNotFoundError, json.JSONDecodeError) as e: print(f"Error reading {file_path}: {str(e)}") continue return f"There's no saved information related to paper {paper_id}." @mcp.resource("papers://folders") def get_available_folders() -> str: """ List all available topic folders in the papers directory. This resource provides a simple list of all available topic folders. """ folders = [] # Get all topic directories if os.path.exists(PAPER_DIR): for topic_dir in os.listdir(PAPER_DIR): topic_path = os.path.join(PAPER_DIR, topic_dir) if os.path.isdir(topic_path): papers_file = os.path.join(topic_path, "papers_info.json") if os.path.exists(papers_file): folders.append(topic_dir) # Create a simple markdown list content = "# Available Topics\n\n" if folders: for folder in folders: content += f"- {folder}\n" content += f"\nUse @{folder} to access papers in that topic.\n" else: content += "No topics found.\n" return content @mcp.resource("papers://{topic}") def get_topic_papers(topic: str) -> str: """ Get detailed information about papers on a specific topic. Args: topic: The research topic to retrieve papers for """ topic_dir = topic.lower().replace(" ", "_") papers_file = os.path.join(PAPER_DIR, topic_dir, "papers_info.json") if not os.path.exists(papers_file): return f"# No papers found for topic: {topic}\n\nTry searching for papers on this topic first." try: with open(papers_file, 'r') as f: papers_data = json.load(f) # Create markdown content with paper details content = f"# Papers on {topic.replace('_', ' ').title()}\n\n" content += f"Total papers: {len(papers_data)}\n\n" for paper_id, paper_info in papers_data.items(): content += f"## {paper_info['title']}\n" content += f"- **Paper ID**: {paper_id}\n" content += f"- **Authors**: {', '.join(paper_info['authors'])}\n" content += f"- **Published**: {paper_info['published']}\n" content += f"- **PDF URL**: [{paper_info['pdf_url']}]({paper_info['pdf_url']})\n\n" content += f"### Summary\n{paper_info['summary'][:500]}...\n\n" content += "---\n\n" return content except json.JSONDecodeError: return f"# Error reading papers data for {topic}\n\nThe papers data file is corrupted." @mcp.prompt() def generate_search_prompt(topic: str, num_papers: int = 5) -> str: """Generate a prompt for Claude to find and discuss academic papers on a specific topic.""" return f"""Search for {num_papers} academic papers about '{topic}' using the search_papers tool. Follow these instructions: 1. First, search for papers using search_papers(topic='{topic}', max_results={num_papers}) 2. For each paper found, extract and organize the following information: - Paper title - Authors - Publication date - Brief summary of the key findings - Main contributions or innovations - Methodologies used - Relevance to the topic '{topic}' 3. Provide a comprehensive summary that includes: - Overview of the current state of research in '{topic}' - Common themes and trends across the papers - Key research gaps or areas for future investigation - Most impactful or influential papers in this area 4. Organize your findings in a clear, structured format with headings and bullet points for easy readability. Please present both detailed information about each paper and a high-level synthesis of the research landscape in {topic}.""" if __name__ == "__main__": # Initialize and run the server mcp.run(transport='sse') Reference DeepLearning.AI - MCP: Build Rich-Context AI Apps with Anthropic ...

Content Introduction Premises Beliefs Focus Change Goal Setting Perfection Health Relationship Self-esteem Final Introduction Extraordinary successful students of Harvard MBAs They really believe in themselves. The sense of confidence. They thought they could do well. They were driven. They were motivated. They thought “I’m going to make it, I’m going to succeed.” They were always asking questions. Always asking questions, initially of their boss, later of their employees, of their partner, children, parents, friends, they were always asking question. They are always at the state of curiosity, always looking up, opening up, wanting to understand the world more. They didn’t say “Now I have my MBA. That’s it. I know enough.” They were life-long learners. They were always asking questions. Change Change is the most important thing. Do work your talk. Give a due time to write down your change, so that you need to change before that time. ...

目录 基本概念 ACID 单对象与多对象操作 弱隔离级别 已提交读（Read Committed） 快照隔离和可重复读（Snapshot Isolation and Repeatable Read） 防止丢失更新（Preventing Lost Updates） 写偏差和幻读（Write Skew and Phantoms） 可序列化（Serializability） 真的串行执行（Actual Serial Execution） 两阶段锁（2PL, Two-Phase Locking） 可序列化快照隔离（SSI, Serializable Snapshot Isolation） 小结 附录：常见事务并发问题的基本概念整理 Reference 基本概念 数据库系统可能出现各种错误，包括软硬件故障，网络故障，并发故障等等。我们需要保证发生这些故障时数据的正确性。 数十年来，事务一直是简化这些问题的首选机制。 事务是将多个读写操作组合成一个逻辑单元的一种方式，整个事务被视作单个操作来进行，要么全部成功（commit），要么全部失败（abort，rollback）。 这样一来，事务就给予了我们一定程度上的安全保证，使得我们不用担心部分失败的情况，以及部分并发问题。 ACID 原子性（Atomicity） 事务的所有操作要么全部成功，要么全部失败。 一致性（Consistency） 对数据的一组特定陈述必须始终成立。 例如，在会计系统中，所有账户整体上必须借贷相抵。 隔离性（Isolation） 同时执行的事务是相互隔离的，它们不能互相影响。 隔离性用于防止并发问题。 持久性（Durability） 一旦事务提交，其所做的修改会永远保存到数据库中。 综合理解 原子性和隔离性保证了一致性。 持久性保证数据不会丢失。 单对象与多对象操作 概念 单对象操作是指在一个事务中只对单个对象进行了操作，多对象操作是指在一个事务中对数据库中的多个对象进行了操作。 单对象写入 可能发生的错误：丢失更新（lost update）。 因此需要保证原子性和隔离性。其中原子性可以通过undo日志来保证，隔离性可以通过给对象上锁来保证。 一些数据库也提供更复杂的原子操作，比如自增操作，这样就不需要“读取-修改-写入”的操作序列了。另一种解决办法是使用CAS操作。 多对象事务 有些情况下一个事务需要对多个对象进行读写操作。 例如一个邮件系统，emails表存储邮件，mailboxs表存储用户的未读邮件数量。当用户收到新邮件时，可能会产生下面的错误（脏读）。 错误处理与中止 错误处理与中止（abort）保证了事务的原子性。如果发生错误，事务可以中止并被安全地重试。 然而，错误处理与中止机制也存在它的问题： 如果事务实际上成功了，但是在服务器试图向客户端确认提交成功时网络发生故障（所以客户端认为提交失败了），那么重试事务会导致事务被执行两次，除非你有一个额外的应用级除重机制。 如果错误是由于负载过大造成的，则重试事务将使问题变得更糟，而不是更好。为了避免这种负反馈循环，可以限制重试次数，使用指数退避算法，并单独处理与过载相关的错误（如果允许）。 仅在临时性错误（例如，由于死锁，异常情况，临时性网络中断和故障切换）后才值得重试。在发生永久性错误（例如，违反约束）之后重试是毫无意义的。 如果事务在数据库之外也有副作用，即使事务被中止，也可能发生这些副作用。例如，副作用是发送电子邮件，那你肯定不希望每次重试事务时都重新发送电子邮件。如果你想确保几个不同的系统一起提交或放弃，可以使用两阶段提交（2PC, two-phase commit）。 如果客户端进程在重试中失效，任何试图写入数据库的数据都将丢失。 弱隔离级别 数据库提供事务隔离（transaction isolation）来隐藏应用程序开发者的并发问题。但是，隔离级别越高，系统性能就越低，所以我们通常使用较弱的隔离级别来防止一部分，而不是全部的并发问题。 ...

This blog is about how I build my blog. The blog is built with Hugo and hosted in GitHub Pages. If you are interested in building your own site using Hugo and GitHub Pages, focus on the Quick Start part. Follow the given references and you can make it. Other parts are about my solutions to some problems or special requirements I faced when I built it. Quick Start Just follow Hugo - Quick Start and Hugo - Host on GitHub. ...

A 5-tuple $(M,C,K,E_k,D_k)$, where $M$: plaintext space $C$: ciphertext space $K$: key space $E_k$: Encryption transformation $D_k$: Decryption transformation Attacks Ciphertext-only attack: only know ciphertext $c$. Known-plaintext attack: know ciphertext-plaintext pair $(c,m)$. Security Requirements $E_k$ and $D_k$ are known to all. It should be computationally infeasible to determine $m$, given $c$. It should be computationally infeasible to determine $D_k$ and $k$, given $c$ and $m$. Transposition Ciphers Let $f$ be a permutation of $Z_d$. ...