Grok MCP

Instructions

Implementation Reference

• src/server.py:480-588 (handler)

  The primary handler implementation for the 'stateful_chat' MCP tool. This async function handles stateful conversations via xAI's stateful API (/v1/responses), managing new and continued chats server-side. The @mcp.tool() decorator handles both registration and schema inference from the type annotations and docstring.
  @mcp.tool() async def stateful_chat( prompt: str, response_id: Optional[str] = None, model: str = "grok-4", system_prompt: Optional[str] = None, include_reasoning: bool = False, temperature: Optional[float] = None, max_tokens: Optional[int] = None ) -> Dict[str, Any]: """ Have ongoing conversations that are saved on xAI's servers for up to 30 days. Unlike regular chat, this maintains conversation history server-side, so you can continue conversations across sessions without managing history yourself. Start a new conversation without a response_id, then use the returned ID to continue it later. Super useful for long-running projects or when you want to pick up where you left off days later. Args: prompt: What you want to say in this turn of the conversation response_id: ID from a previous response to continue that conversation (optional) model: Which Grok model to use (default is grok-4) system_prompt: Instructions for the AI (only used when starting new conversation) include_reasoning: Get a summary of the model's thinking process (default False) temperature: Controls creativity 0-2 max_tokens: Maximum length of response Returns a dict with 'content' (the response), 'response_id' (save this to continue later!), 'status', 'model', 'usage', 'stored_until' (expiration date), and optionally 'reasoning' if you requested it. """ input_messages = [] # System prompt only for new conversations (not when continuing) if system_prompt and not response_id: input_messages.append({ "role": "system", "content": system_prompt }) input_messages.append({ "role": "user", "content": prompt }) request_data: Dict[str, Any] = { "model": model, "input": input_messages, "store": True } if response_id: request_data["previous_response_id"] = response_id # Optional parameters if temperature is not None: request_data["temperature"] = temperature if max_tokens is not None: request_data["max_output_tokens"] = max_tokens if include_reasoning: request_data["reasoning"] = { "include": ["encrypted_content"] } timeout = get_model_timeout(model) client = create_client(timeout=timeout, use_state=True) response = await client.post("/v1/responses", json=request_data) response.raise_for_status() data = response.json() await client.aclose() output = data.get("output", []) content = "" reasoning_summary = None for item in output: if item.get("type") == "message" and item.get("role") == "assistant": for content_item in item.get("content", []): if content_item.get("type") == "output_text": content = content_item.get("text", "") break elif item.get("type") == "reasoning": for summary_item in item.get("summary", []): if summary_item.get("type") == "summary_text": reasoning_summary = summary_item.get("text", "") break expiration = datetime.now() + timedelta(days=30) result = { "content": content, "response_id": data.get("id"), "status": data.get("status"), "model": data.get("model"), "usage": data.get("usage", {}), "stored_until": expiration.strftime("%Y-%m-%d"), "continued_from": response_id if response_id else None } if reasoning_summary: result["reasoning"] = reasoning_summary return result