1- from typing import Annotated , Any , Optional
2- import os
31import json
2+ import os
3+ from typing import Annotated , Any
44
55from fastmcp import Context , FastMCP
66
1010# Optional imports for existing functionality
1111try :
1212 from codegen .cli .api .client import RestAPI
13- from codegen .sdk .core .codebase import Codebase
1413 from codegen .shared .enums .programming_language import ProgrammingLanguage
14+
1515 LEGACY_IMPORTS_AVAILABLE = True
1616except ImportError :
1717 LEGACY_IMPORTS_AVAILABLE = False
2121 from codegen_api_client import ApiClient , Configuration
2222 from codegen_api_client .api import AgentsApi , OrganizationsApi , UsersApi
2323 from codegen_api_client .models import CreateAgentRunInput
24+
2425 API_CLIENT_AVAILABLE = True
2526except ImportError :
2627 API_CLIENT_AVAILABLE = False
2728
2829# Initialize FastMCP server
2930mcp = FastMCP (
30- "codegen-mcp" ,
31- instructions = "MCP server for the Codegen platform. Use the tools and resources to interact with Codegen APIs, setup codegen in your environment, and create/improve Codegen Codemods."
31+ "codegen-mcp" ,
32+ instructions = "MCP server for the Codegen platform. Use the tools and resources to interact with Codegen APIs, setup codegen in your environment, and create/improve Codegen Codemods." ,
3233)
3334
3435# Global API client instances
4142def get_api_client ():
4243 """Get or create the API client instance."""
4344 global _api_client , _agents_api , _organizations_api , _users_api
44-
45+
4546 if not API_CLIENT_AVAILABLE :
46- raise RuntimeError ("codegen-api-client is not available" )
47-
47+ msg = "codegen-api-client is not available"
48+ raise RuntimeError (msg )
49+
4850 if _api_client is None :
4951 # Configure the API client
5052 configuration = Configuration ()
51-
53+
5254 # Set base URL from environment or use default
5355 base_url = os .getenv ("CODEGEN_API_BASE_URL" , "https://api.codegen.com" )
5456 configuration .host = base_url
55-
57+
5658 # Set authentication
5759 api_key = os .getenv ("CODEGEN_API_KEY" )
5860 if api_key :
5961 configuration .api_key = {"Authorization" : f"Bearer { api_key } " }
60-
62+
6163 _api_client = ApiClient (configuration )
6264 _agents_api = AgentsApi (_api_client )
6365 _organizations_api = OrganizationsApi (_api_client )
6466 _users_api = UsersApi (_api_client )
65-
67+
6668 return _api_client , _agents_api , _organizations_api , _users_api
6769
70+
6871# ----- RESOURCES -----
6972
7073
@@ -95,6 +98,7 @@ def get_service_config() -> dict[str, Any]:
9598
9699# Legacy SDK tool (only available if imports are available)
97100if LEGACY_IMPORTS_AVAILABLE :
101+
98102 @mcp .tool ()
99103 def ask_codegen_sdk (query : Annotated [str , "Ask a question to an expert agent for details about any aspect of the codegen sdk core set of classes and utilities" ]):
100104 """Ask questions about the Codegen SDK (requires legacy imports)."""
@@ -118,6 +122,7 @@ def generate_codemod(
118122
119123# Legacy improve codemod tool (only available if imports are available)
120124if LEGACY_IMPORTS_AVAILABLE :
125+
121126 @mcp .tool ()
122127 def improve_codemod (
123128 codemod_source : Annotated [str , "The source code of the codemod to improve" ],
@@ -138,40 +143,37 @@ def improve_codemod(
138143
139144# ----- CODEGEN API TOOLS -----
140145
146+
141147@mcp .tool ()
142148def create_agent_run (
143149 org_id : Annotated [int , "Organization ID" ],
144150 prompt : Annotated [str , "The prompt/task for the agent to execute" ],
145- repo_name : Annotated [Optional [ str ] , "Repository name (optional)" ] = None ,
146- branch_name : Annotated [Optional [ str ] , "Branch name (optional)" ] = None ,
151+ repo_name : Annotated [str | None , "Repository name (optional)" ] = None ,
152+ branch_name : Annotated [str | None , "Branch name (optional)" ] = None ,
147153 ctx : Context = None ,
148154) -> str :
149155 """Create a new agent run in the specified organization."""
150156 try :
151157 _ , agents_api , _ , _ = get_api_client ()
152-
158+
153159 # Create the input object
154- agent_input = CreateAgentRunInput (
155- prompt = prompt ,
156- repo_name = repo_name ,
157- branch_name = branch_name
158- )
159-
160+ agent_input = CreateAgentRunInput (prompt = prompt , repo_name = repo_name , branch_name = branch_name )
161+
160162 # Make the API call
161- response = agents_api .create_agent_run_v1_organizations_org_id_agent_run_post (
162- org_id = org_id ,
163- create_agent_run_input = agent_input
163+ response = agents_api .create_agent_run_v1_organizations_org_id_agent_run_post (org_id = org_id , create_agent_run_input = agent_input )
164+
165+ return json .dumps (
166+ {
167+ "id" : response .id ,
168+ "status" : response .status ,
169+ "created_at" : response .created_at .isoformat () if response .created_at else None ,
170+ "prompt" : response .prompt ,
171+ "repo_name" : response .repo_name ,
172+ "branch_name" : response .branch_name ,
173+ },
174+ indent = 2 ,
164175 )
165-
166- return json .dumps ({
167- "id" : response .id ,
168- "status" : response .status ,
169- "created_at" : response .created_at .isoformat () if response .created_at else None ,
170- "prompt" : response .prompt ,
171- "repo_name" : response .repo_name ,
172- "branch_name" : response .branch_name
173- }, indent = 2 )
174-
176+
175177 except Exception as e :
176178 return f"Error creating agent run: { e } "
177179
@@ -185,23 +187,23 @@ def get_agent_run(
185187 """Get details of a specific agent run."""
186188 try :
187189 _ , agents_api , _ , _ = get_api_client ()
188-
189- response = agents_api .get_agent_run_v1_organizations_org_id_agent_run_agent_run_id_get (
190- org_id = org_id ,
191- agent_run_id = agent_run_id
190+
191+ response = agents_api .get_agent_run_v1_organizations_org_id_agent_run_agent_run_id_get (org_id = org_id , agent_run_id = agent_run_id )
192+
193+ return json .dumps (
194+ {
195+ "id" : response .id ,
196+ "status" : response .status ,
197+ "created_at" : response .created_at .isoformat () if response .created_at else None ,
198+ "updated_at" : response .updated_at .isoformat () if response .updated_at else None ,
199+ "prompt" : response .prompt ,
200+ "repo_name" : response .repo_name ,
201+ "branch_name" : response .branch_name ,
202+ "result" : response .result ,
203+ },
204+ indent = 2 ,
192205 )
193-
194- return json .dumps ({
195- "id" : response .id ,
196- "status" : response .status ,
197- "created_at" : response .created_at .isoformat () if response .created_at else None ,
198- "updated_at" : response .updated_at .isoformat () if response .updated_at else None ,
199- "prompt" : response .prompt ,
200- "repo_name" : response .repo_name ,
201- "branch_name" : response .branch_name ,
202- "result" : response .result
203- }, indent = 2 )
204-
206+
205207 except Exception as e :
206208 return f"Error getting agent run: { e } "
207209
@@ -215,26 +217,16 @@ def get_organizations(
215217 """Get list of organizations the user has access to."""
216218 try :
217219 _ , _ , organizations_api , _ = get_api_client ()
218-
220+
219221 response = organizations_api .get_organizations_v1_organizations_get ()
220-
222+
221223 # Format the response
222224 organizations = []
223225 for org in response .items :
224- organizations .append ({
225- "id" : org .id ,
226- "name" : org .name ,
227- "slug" : org .slug ,
228- "created_at" : org .created_at .isoformat () if org .created_at else None
229- })
230-
231- return json .dumps ({
232- "organizations" : organizations ,
233- "total" : response .total ,
234- "page" : response .page ,
235- "limit" : response .limit
236- }, indent = 2 )
237-
226+ organizations .append ({"id" : org .id , "name" : org .name , "slug" : org .slug , "created_at" : org .created_at .isoformat () if org .created_at else None })
227+
228+ return json .dumps ({"organizations" : organizations , "total" : response .total , "page" : response .page , "limit" : response .limit }, indent = 2 )
229+
238230 except Exception as e :
239231 return f"Error getting organizations: { e } "
240232
@@ -249,28 +241,16 @@ def get_users(
249241 """Get list of users in an organization."""
250242 try :
251243 _ , _ , _ , users_api = get_api_client ()
252-
253- response = users_api .get_users_v1_organizations_org_id_users_get (
254- org_id = org_id
255- )
256-
244+
245+ response = users_api .get_users_v1_organizations_org_id_users_get (org_id = org_id )
246+
257247 # Format the response
258248 users = []
259249 for user in response .items :
260- users .append ({
261- "id" : user .id ,
262- "email" : user .email ,
263- "name" : user .name ,
264- "created_at" : user .created_at .isoformat () if user .created_at else None
265- })
266-
267- return json .dumps ({
268- "users" : users ,
269- "total" : response .total ,
270- "page" : response .page ,
271- "limit" : response .limit
272- }, indent = 2 )
273-
250+ users .append ({"id" : user .id , "email" : user .email , "name" : user .name , "created_at" : user .created_at .isoformat () if user .created_at else None })
251+
252+ return json .dumps ({"users" : users , "total" : response .total , "page" : response .page , "limit" : response .limit }, indent = 2 )
253+
274254 except Exception as e :
275255 return f"Error getting users: { e } "
276256
@@ -284,20 +264,20 @@ def get_user(
284264 """Get details of a specific user in an organization."""
285265 try :
286266 _ , _ , _ , users_api = get_api_client ()
287-
288- response = users_api .get_user_v1_organizations_org_id_users_user_id_get (
289- org_id = org_id ,
290- user_id = user_id
267+
268+ response = users_api .get_user_v1_organizations_org_id_users_user_id_get (org_id = org_id , user_id = user_id )
269+
270+ return json .dumps (
271+ {
272+ "id" : response .id ,
273+ "email" : response .email ,
274+ "name" : response .name ,
275+ "created_at" : response .created_at .isoformat () if response .created_at else None ,
276+ "updated_at" : response .updated_at .isoformat () if response .updated_at else None ,
277+ },
278+ indent = 2 ,
291279 )
292-
293- return json .dumps ({
294- "id" : response .id ,
295- "email" : response .email ,
296- "name" : response .name ,
297- "created_at" : response .created_at .isoformat () if response .created_at else None ,
298- "updated_at" : response .updated_at .isoformat () if response .updated_at else None
299- }, indent = 2 )
300-
280+
301281 except Exception as e :
302282 return f"Error getting user: { e } "
303283
@@ -314,7 +294,8 @@ def run_server(transport: str = "stdio", host: str = "localhost", port: int | No
314294 print (f"HTTP transport not yet implemented. Would run on { host } :{ port } " )
315295 mcp .run (transport = "stdio" ) # Fallback to stdio for now
316296 else :
317- raise ValueError (f"Unsupported transport: { transport } " )
297+ msg = f"Unsupported transport: { transport } "
298+ raise ValueError (msg )
318299
319300
320301if __name__ == "__main__" :
0 commit comments