Zenn MCP Server

Instructions

Implementation Reference

• src/mcp_server_zenn/server.py:228-231 (handler)

  The specific handler function for the 'book' tool. It parses the input arguments into a Book model instance and fetches the books data via fetch_books.

  async def handle_books(arguments: dict) -> dict:
      query = Book.from_arguments(arguments)
      return await fetch_books(query)

• src/mcp_server_zenn/server.py:142-205 (schema)

  Pydantic model defining the input schema for the 'book' tool, including parsing from arguments, query param conversion, and the Tool definition with inputSchema.

  class Book(BaseModel):
      """Fetch books from Zenn.dev"""
  
      model_config = ConfigDict(
          validate_assignment=True,
          frozen=True,
          extra="forbid",
      )
  
      username: Optional[str] = Field(default=None, description="Username of the book author")
      topicname: Optional[str] = Field(default=None, description="Topic name of the book")
      order: Optional[Order] = Field(
          default=Order.LATEST,
          description=f"Order of the books. Choose from {Order.LATEST.value} or {Order.OLDEST.value}. Default: {Order.LATEST.value}",
      )
      page: Optional[int] = Field(default=1, description="Page number of the books. Default: 1")
      count: Optional[int] = Field(default=48, description="Number of books per page. Default: 48")
  
      @staticmethod
      def from_arguments(arguments: dict) -> "Book":
          return Book(
              username=arguments.get("username"),
              topicname=arguments.get("topicname"),
              order=Order.from_str(arguments.get("order", Order.LATEST.value)),
              page=arguments.get("page", 1),
              count=arguments.get("count", 48),
          )
  
      def to_query_param(self) -> dict:
          param = {}
          if self.username:
              param["username"] = self.username.lower()
          if self.topicname:
              param["topicname"] = self.topicname.lower()
          if self.order:
              param["order"] = self.order.value
          if self.page:
              param["page"] = self.page
          if self.count:
              param["count"] = self.count
          return param
  
      @staticmethod
      def tool() -> Tool:
          return Tool(
              name=ZennTool.BOOK.value,
              description="Fetch books from Zenn.dev",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "username": {"type": "string", "description": Book.model_fields["username"].description},
                      "topicname": {"type": "string", "description": Book.model_fields["topicname"].description},
                      "order": {
                          "type": "string",
                          "description": Book.model_fields["order"].description,
                          "enum": [Order.LATEST.value, Order.OLDEST.value],
                      },
                      "page": {"type": "integer", "description": Book.model_fields["page"].description},
                      "count": {"type": "integer", "description": Book.model_fields["count"].description},
                  },
                  "required": [],
              },
          )

• src/mcp_server_zenn/server.py:243-245 (registration)

  Registers the 'book' tool (via Book.tool()) with the MCP server.

  @server.list_tools()
  async def list_tools() -> list[Tool]:
      return [Article.tool(), Book.tool()]

• src/mcp_server_zenn/server.py:219-220 (helper)

  Helper function that performs the HTTP request to fetch books from Zenn API using the query parameters from the Book model.

  async def fetch_books(query: Book) -> dict:
      return await request(URLResource.BOOKS, query.to_query_param())

• src/mcp_server_zenn/server.py:207-212 (helper)

  Generic helper to make HTTP GET request to Zenn API endpoints and return JSON response.

  async def request(resource: URLResource, query: dict) -> dict:
      url = f"{BASE_URL}{resource.value}"
      async with httpx.AsyncClient() as client:
          response = await client.get(url, params=query)
          response.raise_for_status()
          return response.json()